PHP は SWOOLE 拡張機能を使用してタイミング同期を実装します

この記事では、PHP SWOOLE を使用して非同期スケジュールされたタスク システムを作成する方法を紹介します。

Nanning Company では、呼び出しシステムを使用しています。支店の通話サーバーはイントラネット上にあり、技術的な手段でマッピングされているため、支店と南寧間のネットワークが不安定であるため、通話データを分析する必要があります。南寧と同期します。

最も簡単な方法は、MySQL のマスター/スレーブ同期を直接設定して、データを Nanning に同期することです。しかし、営業電話システムの会社は私たちに MySQL 権限を与えません。したがって、この方法は放棄するしかありません。

そこで、PHP を使用して単純な PHP タイミング同期ツールを実装することを考えました。その後、PHP プロセスがバックグラウンドで実行されるため、最初に PHP コンポーネントである SWOOLE にたどり着きました。議論の結果、The Branch Company が設立しました。半日で生成されるデータの最大量は約 5,000 であるため、このソリューションは実現可能ですので、ぜひ実行してください。

PHP SWOOLE を使用して、非同期のスケジュールされたタスク システムを作成します。

MySQL データベース自体のマスター/スレーブ同期は、マスター データベース内のバイナリ ログを解析することによって、データをスレーブ データベースに同期します。ただし、PHP を使用してデータを同期する場合、マスター ライブラリからデータをバッチでクエリし、それを南寧のスレーブ ライブラリに挿入することしかできません。

ここで使用するフレームワークは ThinkPHP 3.2 です。

最初に PHP 拡張機能 SWOOLE をインストールします。特別な関数は使用しないため、ここでは簡単にインストールするために pecl を使用します。 :

pecl install swoole

インストールが完了したら、extension="swoole.so" を php.ini に追加します。インストールが完了したら、phpinfo を使用します() 成功したかどうかを確認します。

#インストールは成功しました。ビジネスを書きましょう。

#サーバー

1. まず、バックグラウンド サーバーを起動し、ポート 9501

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 でリッスンします。タスクの実行、データのクエリとマスターからの書き込みが行われます。ライブラリをスレーブ データベースにコピーします。

public function onReceive($serv, $fd, $from_id, $data)
{
 //使用json_decode 解析任务数据
 $areas = json_decode($data,true);
 foreach ($areas as $area){
  //投递异步任务
  $serv->task($area);
 }
}

4. タスクが完了したら、

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");
}

クライアント プッシュ タスク

を呼び出します。これで基本的には完了です。次にクライアント タスクを書き込みます。

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

のプッシュは基本的に完了します。残りは、定期的に実行するためのシェル スクリプトを書きます: /home/wwwroot/sync_db/crontab/send.sh

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;
}

Using crontab スケジュールされたタスクを使用するには、スクリプトをスケジュールされたタスクに追加します

#!/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

ヒント: タスクのプッシュと実行が成功したかどうかを確認できるように、ログ書き込み操作をスクリプトに追加することをお勧めします。

これで基本的には完成しました。プログラムを最適化する必要があります~~~。もっと良い方法があれば、ぜひ提案してください。

上記がこの記事の全内容です。その他の関連コンテンツについては、PHP 中国語 Web サイトをご覧ください。

関連する推奨事項:

PHP は、2 つの配列内の異なる要素をクエリするメソッドを実装します。

php は、mysql 接続プールの効果を実装します

以上がPHP は SWOOLE 拡張機能を使用してタイミング同期を実装しますの詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。