PHP使用SWOOLE扩展实现定时同步 MySQL 数据

浅浅的花香味﹌ 2022-05-15 14:24 245阅读 0赞

南宁公司和几个分公司之间都使用了呼叫系统，然后现在需要做一个呼叫通话数据分析，由于分公司的呼叫服务器是在内网，通过技术手段映射出来，分公司到南宁之间的网络不稳定，所以需要把分公司的通话数据同步到南宁。

本身最简单的方法就是直接配置MySQL的主从同步就可以同步数据到南宁来了。但是销售呼叫系统那边的公司不给MySQL权限我们。 所以这个方法只能放弃了。

于是我们干脆的想，使用PHP来实现定时一个简易的PHP定时同步工具，然后PHP进程常驻后台运行，所以首先就先到了一个PHP组件：SWOOLE，经过讨论，分公司的每天半天生成的数据量最大在5000条左右，所以这个方案是可行，就这样干。

我们使用PHP SWOOLE 做一个异步的定时任务系统。

本身MySQL数据库的主从同步是通过解析Master库中的binary-log来进行同步数据到从库的。然而我们使用PHP来同步数据的时候，那么只能从master库分批查询数据，然后插入到南宁的slave库来了。

这里我们使用的框架是 `ThinkPHP 3.2` .

首先安装PHP扩展： SWOOLE,因为没有使用到特别的功能，所以这里我们使用pecl来快速安装:

[?][Link 1]

<table> 
 <tbody> 
 <tr> 
 <td> 1 </td> 
 <td> <code>pecl </code><code>install</code> <code>swoole</code> </td> 
 </tr> 
 </tbody> 
</table>

安装完成后在 `php.ini` 里面加入 `extension="swoole.so"` 安装完成后，我们使用 `phpinfo()` 来检查是否成功了.

[![201704090941518.png][]][201704090941518.png]

安装成功了，我们就来写业务.

**服务端**

1、首先启动一个后台的服务端,监听端口9501

[?][Link 1]

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 </td> 
 <td> <code>public</code> <code>function</code> <code>index()</code> <code>{ </code> <code>&nbsp;</code><code>$serv</code> <code>= </code><code>new</code> <code>\swoole_server(</code><code>"0.0.0.0"</code><code>, 9501);</code> <code>&nbsp;</code><code>$serv</code><code>-&gt;set([</code> <code>&nbsp;&nbsp;</code><code>'worker_num'</code> <code>=&gt; 1,</code><code>//一般设置为服务器CPU数的1-4倍</code> <code>&nbsp;&nbsp;</code><code>'task_worker_num'</code> <code>=&gt; 8,</code><code>//task进程的数量</code> <code>&nbsp;&nbsp;</code><code>'daemonize'</code> <code>=&gt; 1,</code><code>//以守护进程执行</code> <code>&nbsp;&nbsp;</code><code>'max_request'</code> <code>=&gt; 10000,</code><code>//最大请求数量</code> <code>&nbsp;&nbsp;</code><code>"task_ipc_mode "</code> <code>=&gt; 2 </code><code>//使用消息队列通信，并设置为争抢模式</code> <code>&nbsp;</code><code>]);</code> <code>&nbsp;</code><code>$serv</code><code>-&gt;on(</code><code>'Receive'</code><code>, [</code><code>$this</code><code>, </code><code>'onReceive'</code><code>]);</code><code>//接收任务，并投递</code> <code>&nbsp;</code><code>$serv</code><code>-&gt;on(</code><code>'Task'</code><code>, [</code><code>$this</code><code>, </code><code>'onTask'</code><code>]);</code><code>//可以在这个方法里面处理任务</code> <code>&nbsp;</code><code>$serv</code><code>-&gt;on(</code><code>'Finish'</code><code>, [</code><code>$this</code><code>, </code><code>'onFinish'</code><code>]);</code><code>//任务完成时候调用</code> <code>&nbsp;</code><code>$serv</code><code>-&gt;start();</code> <code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

2、接收和投递任务

[?][Link 1]

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 </td> 
 <td> <code>public</code> <code>function</code> <code>onReceive(</code><code>$serv</code><code>, </code><code>$fd</code><code>, </code><code>$from_id</code><code>, </code><code>$data</code><code>)</code> <code>{ </code> <code>&nbsp;</code><code>//使用json_decode 解析任务数据</code> <code>&nbsp;</code><code>$areas</code> <code>= json_decode(</code><code>$data</code><code>,true);</code> <code>&nbsp;</code><code>foreach</code> <code>(</code><code>$areas</code> <code>as</code> <code>$area</code><code>){ </code> <code>&nbsp;&nbsp;</code><code>//投递异步任务</code> <code>&nbsp;&nbsp;</code><code>$serv</code><code>-&gt;task(</code><code>$area</code><code>);</code> <code>&nbsp;</code><code>}</code> <code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

3、任务执行，数据从master库查询和写入到slave数据库

[?][Link 1]

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 </td> 
 <td> <code>public</code> <code>function</code> <code>onTask(</code><code>$serv</code><code>, </code><code>$task_id</code><code>, </code><code>$from_id</code><code>, </code><code>$task_data</code><code>)</code> <code>{ </code> <code>&nbsp;</code><code>$area</code> <code>= </code><code>$task_data</code><code>;</code><code>//参数是地区编号</code> <code>&nbsp;</code><code>$rows</code> <code>= 50; </code><code>//每页多少条</code> <code>&nbsp;</code><code>//主库地址,根据参数地区（$area）编号切换master数据库连接</code> <code>&nbsp;</code><code>//从库MySQL实例,根据参数地区（$area）编号切换slave数据库连接</code> <code>&nbsp;</code><code>//由于程序是常驻内存的，所以MySQL连接可以使用长连接，然后重复利用。要使用设计模式的，可以使用对象池模式</code> <code>&nbsp;</code><code>Code......</code> &nbsp; <code>&nbsp;</code><code>//master 库为分公司的数据库，slave库为数据同步到南宁后的从库</code> <code>&nbsp;</code><code>Code......</code> &nbsp; <code>&nbsp;</code><code>//使用$sql获取从库中最大的自增: SELECT MAX(id) AS maxid FROM ss_cdr_cdr_info limit 1</code> <code>&nbsp;</code><code>$slaveMaxIncrementId</code> <code>= ...;</code> &nbsp; <code>&nbsp;</code><code>//使用$sql获取主库中最大的自增: SELECT MAX(id) AS maxid FROM ss_cdr_cdr_info limit 1</code> <code>&nbsp;</code><code>$masterMaxIncrementId</code> <code>= ...;</code> &nbsp; <code>&nbsp;</code><code>//如果相等的就不同步了</code> <code>&nbsp;</code><code>if</code><code>(</code><code>$slaveMaxIncrementId</code> <code>&gt;= </code><code>$masterMaxIncrementId</code><code>){ </code> <code>&nbsp;&nbsp;</code><code>return</code> <code>false;</code> <code>&nbsp;</code><code>}</code> &nbsp; <code>&nbsp;</code><code>//根据条数计算页数</code> <code>&nbsp;</code><code>$dataNumber</code> <code>= </code><code>ceil</code><code>(</code><code>$masterMaxIncrementId</code> <code>- </code><code>$slaveMaxIncrementId</code><code>);</code> <code>&nbsp;</code><code>$eachNumber</code> <code>= </code><code>ceil</code><code>(</code><code>$dataNumber</code> <code>/ </code><code>$rows</code><code>);</code> <code>&nbsp;</code><code>$left</code> <code>= 0;</code> &nbsp; <code>&nbsp;</code><code>//根据页数来进行分批循环进行写入，要记得及时清理内存</code> <code>&nbsp;</code><code>for</code> <code>(</code><code>$i</code> <code>= 0; </code><code>$i</code> <code>&lt; </code><code>$eachNumber</code><code>; </code><code>$i</code><code>++) { </code> <code>&nbsp;&nbsp;</code><code>$left</code> <code>= </code><code>$i</code> <code>== 0 ? </code><code>$slaveMaxIncrementId</code> <code>: </code><code>$left</code> <code>+ </code><code>$rows</code><code>;</code> <code>&nbsp;&nbsp;</code><code>$right</code> <code>= </code><code>$left</code> <code>+ </code><code>$rows</code><code>;</code> <code>&nbsp;&nbsp;</code><code>//生成分批查询条件</code> <code>&nbsp;&nbsp;</code><code>//$where = "id &gt; $left AND &lt;= $right";</code> <code>&nbsp;&nbsp;</code><code>$masterData</code> <code>= ...;</code><code>//从主库查询数据</code> <code>&nbsp;&nbsp;</code><code>$slaveLastInsertId</code> <code>= ...;</code><code>//插入到从库</code> <code>&nbsp;&nbsp;</code><code>unset(</code><code>$masterData</code><code>,</code><code>$slaveLastInsertId</code><code>);</code> <code>&nbsp;</code><code>}</code> &nbsp; <code>&nbsp;</code><code>echo</code> <code>"New AsyncTask[id=$task_id]"</code><code>.PHP_EOL;</code> <code>&nbsp;</code><code>$serv</code><code>-&gt;finish(</code><code>"$area -&gt; OK"</code><code>);</code> <code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

4、任务完成时候调用

[?][Link 1]

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 </td> 
 <td> <code>public</code> <code>function</code> <code>onFinish(</code><code>$serv</code><code>, </code><code>$task_id</code><code>, </code><code>$task_data</code><code>)</code> <code>{ </code> <code>&nbsp;</code><code>echo</code> <code>"AsyncTask[$task_id] Finish: $task_data"</code><code>.PHP_EOL;</code> <code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

**客户端推送任务**

到此基本完成，剩下来我们来写客户端任务推送

[?][Link 1]

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 </td> 
 <td> <code>public</code> <code>function</code> <code>index()</code> <code>{ </code> <code>&nbsp;</code><code>$client</code> <code>= </code><code>new</code> <code>\swoole_client(SWOOLE_SOCK_TCP);</code> <code>&nbsp;</code><code>if</code> <code>(!</code><code>$client</code><code>-&gt;connect(</code><code>'127.0.0.1'</code><code>, 9501, 1)) { </code> <code>&nbsp;&nbsp;</code><code>throw</code> <code>new</code> <code>Exception(</code><code>'链接SWOOLE服务错误'</code><code>);</code> <code>&nbsp;</code><code>}</code> <code>&nbsp;</code><code>$areas</code> <code>= json_encode([</code><code>'liuzhou'</code><code>,</code><code>'yulin'</code><code>,</code><code>'beihai'</code><code>,</code><code>'guilin'</code><code>]);</code> <code>&nbsp;</code><code>//开始遍历检查</code> <code>&nbsp;</code><code>$client</code><code>-&gt;send(</code><code>$areas</code><code>);</code> <code>&nbsp;</code><code>echo</code> <code>"任务发送成功"</code><code>.PHP_EOL;</code> <code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

至此基本完成了，剩下的我们来写一个shell脚本定时执行：/home/wwwroot/sync\_db/crontab/send.sh

[?][Link 1]

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 </td> 
 <td> <code>#!/bin/bash</code> <code>PATH=</code><code>/bin</code><code>:</code><code>/sbin</code><code>:</code><code>/usr/bin</code><code>:</code><code>/usr/sbin</code><code>:</code><code>/usr/local/bin</code><code>:</code><code>/usr/local/sbin</code><code>:~</code><code>/bin</code> <code>export</code> <code>PATH</code> &nbsp; <code># 定时推送异步的数据同步任务</code> <code>/usr/bin/php</code> <code>/home/wwwroot/sync_db/server</code><code>.php home</code><code>/index/index</code> </td> 
 </tr> 
 </tbody> 
</table>

使用crontab定时任务，我们把脚本加入定时任务

[?][Link 1]

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 </td> 
 <td> <code>#设置每天12：30执行数据同步任务</code> <code>30 12 * * * root </code><code>/home/wwwroot/sync_db/crontab/send</code><code>.sh</code> <code>#设置每天19：00执行数据同步任务</code> <code>0 19 * * * root </code><code>/home/wwwroot/sync_db/crontab/send</code><code>.sh</code> </td> 
 </tr> 
 </tbody> 
</table>

Tips: 最好推荐在里面加入写日志操作，这样好知道是任务推送、执行是否成功。

至此基本完成，程序有待优化~~~，各位看客有更好的方法欢迎提出。

参考文献：

[https://www.jb51.net/article/110702.htm][https_www.jb51.net_article_110702.htm]

[Link 1]: https://www.jb51.net/article/110702.htm#
[201704090941518.png]: https://files.jb51.net/file_images/article/201704/201704090941518.png
[https_www.jb51.net_article_110702.htm]: https://www.jb51.net/article/110702.htm