PHP使用SWOOLE擴(kuò)展實(shí)現(xiàn)定時(shí)同步 MySQL 數(shù)據(jù)

更新時(shí)間：2017年04月09日 09:47:11 作者：Mr.Cong

本文給大家介紹的是一次比較特殊的任務(wù)，我們使用PHP SWOOLE 做一個(gè)異步的定時(shí)任務(wù)系統(tǒng)，具體如何來(lái)實(shí)現(xiàn)的呢，接下來(lái)來(lái)我們好好看下吧

南寧公司和幾個(gè)分公司之間都使用了呼叫系統(tǒng)，然后現(xiàn)在需要做一個(gè)呼叫通話(huà)數(shù)據(jù)分析，由于分公司的呼叫服務(wù)器是在內(nèi)網(wǎng)，通過(guò)技術(shù)手段映射出來(lái)，分公司到南寧之間的網(wǎng)絡(luò)不穩(wěn)定，所以需要把分公司的通話(huà)數(shù)據(jù)同步到南寧。

本身最簡(jiǎn)單的方法就是直接配置MySQL的主從同步就可以同步數(shù)據(jù)到南寧來(lái)了。但是銷(xiāo)售呼叫系統(tǒng)那邊的公司不給MySQL權(quán)限我們。所以這個(gè)方法只能放棄了。

于是我們干脆的想，使用PHP來(lái)實(shí)現(xiàn)定時(shí)一個(gè)簡(jiǎn)易的PHP定時(shí)同步工具，然后PHP進(jìn)程常駐后臺(tái)運(yùn)行，所以首先就先到了一個(gè)PHP組件：SWOOLE，經(jīng)過(guò)討論，分公司的每天半天生成的數(shù)據(jù)量最大在5000條左右，所以這個(gè)方案是可行，就這樣干。

我們使用PHP SWOOLE 做一個(gè)異步的定時(shí)任務(wù)系統(tǒng)。

本身MySQL數(shù)據(jù)庫(kù)的主從同步是通過(guò)解析Master庫(kù)中的binary-log來(lái)進(jìn)行同步數(shù)據(jù)到從庫(kù)的。然而我們使用PHP來(lái)同步數(shù)據(jù)的時(shí)候，那么只能從master庫(kù)分批查詢(xún)數(shù)據(jù)，然后插入到南寧的slave庫(kù)來(lái)了。

這里我們使用的框架是 ThinkPHP 3.2 .

首先安裝PHP擴(kuò)展： SWOOLE,因?yàn)闆](méi)有使用到特別的功能，所以這里我們使用pecl來(lái)快速安裝:

pecl install swoole

安裝完成后在 php.ini 里面加入 extension="swoole.so" 安裝完成后，我們使用 phpinfo() 來(lái)檢查是否成功了.

安裝成功了，我們就來(lái)寫(xiě)業(yè)務(wù).

服務(wù)端

1、首先啟動(dòng)一個(gè)后臺(tái)的服務(wù)端,監(jiān)聽(tīng)端口9501

public function index()
{
 $serv = new \swoole_server("0.0.0.0", 9501);
 $serv->set([
  'worker_num' => 1,//一般設(shè)置為服務(wù)器CPU數(shù)的1-4倍
  'task_worker_num' => 8,//task進(jìn)程的數(shù)量
  'daemonize' => 1,//以守護(hù)進(jìn)程執(zhí)行
  'max_request' => 10000,//最大請(qǐng)求數(shù)量
  "task_ipc_mode " => 2 //使用消息隊(duì)列通信，并設(shè)置為爭(zhēng)搶模式
 ]);
 $serv->on('Receive', [$this, 'onReceive']);//接收任務(wù)，并投遞
 $serv->on('Task', [$this, 'onTask']);//可以在這個(gè)方法里面處理任務(wù)
 $serv->on('Finish', [$this, 'onFinish']);//任務(wù)完成時(shí)候調(diào)用
 $serv->start();
}

2、接收和投遞任務(wù)

public function onReceive($serv, $fd, $from_id, $data)
{
 //使用json_decode 解析任務(wù)數(shù)據(jù)
 $areas = json_decode($data,true);
 foreach ($areas as $area){
  //投遞異步任務(wù)
  $serv->task($area);
 }
}

3、任務(wù)執(zhí)行，數(shù)據(jù)從master庫(kù)查詢(xún)和寫(xiě)入到slave數(shù)據(jù)庫(kù)

public function onTask($serv, $task_id, $from_id, $task_data)
{
 $area = $task_data;//參數(shù)是地區(qū)編號(hào)
 $rows = 50; //每頁(yè)多少條
 //主庫(kù)地址,根據(jù)參數(shù)地區(qū)（$area）編號(hào)切換master數(shù)據(jù)庫(kù)連接
 //從庫(kù)MySQL實(shí)例,根據(jù)參數(shù)地區(qū)（$area）編號(hào)切換slave數(shù)據(jù)庫(kù)連接
 //由于程序是常駐內(nèi)存的，所以MySQL連接可以使用長(zhǎng)連接，然后重復(fù)利用。要使用設(shè)計(jì)模式的，可以使用對(duì)象池模式
 Code......

 //master 庫(kù)為分公司的數(shù)據(jù)庫(kù)，slave庫(kù)為數(shù)據(jù)同步到南寧后的從庫(kù)
 Code......

 //使用$sql獲取從庫(kù)中最大的自增: SELECT MAX(id) AS maxid FROM ss_cdr_cdr_info limit 1
 $slaveMaxIncrementId = ...;

 //使用$sql獲取主庫(kù)中最大的自增: SELECT MAX(id) AS maxid FROM ss_cdr_cdr_info limit 1
 $masterMaxIncrementId = ...;

 //如果相等的就不同步了
 if($slaveMaxIncrementId >= $masterMaxIncrementId){
  return false;
 }

 //根據(jù)條數(shù)計(jì)算頁(yè)數(shù)
 $dataNumber = ceil($masterMaxIncrementId - $slaveMaxIncrementId);
 $eachNumber = ceil($dataNumber / $rows);
 $left = 0;

 //根據(jù)頁(yè)數(shù)來(lái)進(jìn)行分批循環(huán)進(jìn)行寫(xiě)入，要記得及時(shí)清理內(nèi)存
 for ($i = 0; $i < $eachNumber; $i++) {
  $left = $i == 0 ? $slaveMaxIncrementId : $left + $rows;
  $right = $left + $rows;
  //生成分批查詢(xún)條件
  //$where = "id > $left AND <= $right";
  $masterData = ...;//從主庫(kù)查詢(xún)數(shù)據(jù)
  $slaveLastInsertId = ...;//插入到從庫(kù)
  unset($masterData,$slaveLastInsertId);
 }

 echo "New AsyncTask[id=$task_id]".PHP_EOL;
 $serv->finish("$area -> OK");
}

4、任務(wù)完成時(shí)候調(diào)用

public function onFinish($serv, $task_id, $task_data)
{
 echo "AsyncTask[$task_id] Finish: $task_data".PHP_EOL;
}

客戶(hù)端推送任務(wù)

到此基本完成，剩下來(lái)我們來(lái)寫(xiě)客戶(hù)端任務(wù)推送

public function index()
{
 $client = new \swoole_client(SWOOLE_SOCK_TCP);
 if (!$client->connect('127.0.0.1', 9501, 1)) {
  throw new Exception('鏈接SWOOLE服務(wù)錯(cuò)誤');
 }
 $areas = json_encode(['liuzhou','yulin','beihai','guilin']);
 //開(kāi)始遍歷檢查
 $client->send($areas);
 echo "任務(wù)發(fā)送成功".PHP_EOL;
}

至此基本完成了，剩下的我們來(lái)寫(xiě)一個(gè)shell腳本定時(shí)執(zhí)行：/home/wwwroot/sync_db/crontab/send.sh

#!/bin/bash
PATH=/bin:/sbin:/usr/bin:/usr/sbin:/usr/local/bin:/usr/local/sbin:~/bin
export PATH

# 定時(shí)推送異步的數(shù)據(jù)同步任務(wù)
/usr/bin/php /home/wwwroot/sync_db/server.php home/index/index

使用crontab定時(shí)任務(wù)，我們把腳本加入定時(shí)任務(wù)

#設(shè)置每天12：30執(zhí)行數(shù)據(jù)同步任務(wù)
30 12 * * * root /home/wwwroot/sync_db/crontab/send.sh
#設(shè)置每天19：00執(zhí)行數(shù)據(jù)同步任務(wù)
0 19 * * * root /home/wwwroot/sync_db/crontab/send.sh

Tips: 最好推薦在里面加入寫(xiě)日志操作，這樣好知道是任務(wù)推送、執(zhí)行是否成功。

至此基本完成，程序有待優(yōu)化~~~，各位看客有更好的方法歡迎提出。

您可能感興趣的文章: