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

时间:2022-05-03 01:25:50

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

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

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

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

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

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

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

?
1
pecl install swoole

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

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

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

服务端

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

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

2、接收和投递任务

?
1
2
3
4
5
6
7
8
9
public function onreceive($serv, $fd, $from_id, $data)
{
 //使用json_decode 解析任务数据
 $areas = json_decode($data,true);
 foreach ($areas as $area){
  //投递异步任务
  $serv->task($area);
 }
}

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

?
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
public function ontask($serv, $task_id, $from_id, $task_data)
{
 $area = $task_data;//参数是地区编号
 $rows = 50; //每页多少条
 //主库地址,根据参数地区($area)编号切换master数据库连接
 //从库mysql实例,根据参数地区($area)编号切换slave数据库连接
 //由于程序是常驻内存的,所以mysql连接可以使用长连接,然后重复利用。要使用设计模式的,可以使用对象池模式
 code......
 
 //master 库为分公司的数据库,slave库为数据同步到南宁后的从库
 code......
 
 //使用$sql获取从库中最大的自增: select max(id) as maxid from ss_cdr_cdr_info limit 1
 $slavemaxincrementid = ...;
 
 //使用$sql获取主库中最大的自增: select max(id) as maxid from ss_cdr_cdr_info limit 1
 $mastermaxincrementid = ...;
 
 //如果相等的就不同步了
 if($slavemaxincrementid >= $mastermaxincrementid){
  return false;
 }
 
 //根据条数计算页数
 $datanumber = ceil($mastermaxincrementid - $slavemaxincrementid);
 $eachnumber = ceil($datanumber / $rows);
 $left = 0;
 
 //根据页数来进行分批循环进行写入,要记得及时清理内存
 for ($i = 0; $i < $eachnumber; $i++) {
  $left = $i == 0 ? $slavemaxincrementid : $left + $rows;
  $right = $left + $rows;
  //生成分批查询条件
  //$where = "id > $left and <= $right";
  $masterdata = ...;//从主库查询数据
  $slavelastinsertid = ...;//插入到从库
  unset($masterdata,$slavelastinsertid);
 }
 
 echo "new asynctask[id=$task_id]".php_eol;
 $serv->finish("$area -> ok");
}

4、任务完成时候调用

?
1
2
3
4
public function onfinish($serv, $task_id, $task_data)
{
 echo "asynctask[$task_id] finish: $task_data".php_eol;
}

客户端推送任务

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

?
1
2
3
4
5
6
7
8
9
10
11
public function index()
{
 $client = new \swoole_client(swoole_sock_tcp);
 if (!$client->connect('127.0.0.1', 9501, 1)) {
  throw new exception('链接swoole服务错误');
 }
 $areas = json_encode(['liuzhou','yulin','beihai','guilin']);
 //开始遍历检查
 $client->send($areas);
 echo "任务发送成功".php_eol;
}

至此基本完成了,剩下的我们来写一个shell脚本定时执行:/home/wwwroot/sync_db/crontab/send.sh

?
1
2
3
4
5
6
#!/bin/bash
path=/bin:/sbin:/usr/bin:/usr/sbin:/usr/local/bin:/usr/local/sbin:~/bin
export path
 
# 定时推送异步的数据同步任务
/usr/bin/php /home/wwwroot/sync_db/server.php home/index/index

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

?
1
2
3
4
#设置每天12:30执行数据同步任务
30 12 * * * root /home/wwwroot/sync_db/crontab/send.sh
#设置每天19:00执行数据同步任务
0 19 * * * root /home/wwwroot/sync_db/crontab/send.sh

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

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

原文链接:https://cong5.net/post/php-swoole-sync-call-data-for-subsidiary-to-nanning