swoole+Redis将实时数据的推送
一 实现功能
设计师订单如果设计师未抢单,超时(5分钟)设计订单时时给设计师派送,
设计师公众号中收到派单信息
设计发布者收到派单成功信息
环境
centos6.10
redis-4.0.2
swoole-src-4.4.12
php-7.1.5
MYsyql5.7
在centos6默认是gcc-4.7,安装swoole的时候需要升级到gcc-4.8
二 实现流程
1.开启swoole server端监听
2.开启swoole client连接执行定时执行
3.使用swoole task 异步执行推送逻辑
开始监听
服务端窗口
# php71 pushServer.php
client连接执行开始任务
客户端窗口
# php71 pushClient.php start
默认start开启5个client tcp链接,每个链接开启一个1s定时器
开启后服务端窗口的变化
[root@111111 swoole_server]# php71 pushServer.php
Client-1: 连接成功
reactor-7 Client-1 接受数据: data=start
Client-1: 连接结束
Client-2: 连接成功
reactor-0 Client-2 接受数据: data=start
Client-3: 连接成功
Client-2: 连接结束
reactor-7 Client-3 接受数据: data=start
Client-3: 连接结束
Client-4: 连接成功
reactor-0 Client-4 接受数据: data=start
Client-4: 连接结束
Client-5: 连接成功
reactor-7 Client-5 接受数据: data=start
Client-5: 连接结束
2019-12-14 01:29:15reactor-7 client-1 timer-2: 定时器第1 次执行
redis添加数据
向order_designer_list队列中添加数据
127.0.0.1:6379> LPUSH order_designer_list 1912136916313##150481##1576140373##oLvqQwo25p5myELvO5VXj0k-7ngk##李伟测试##13926
(integer) 1
127.0.0.1:6379> LRANGE order_designer_list 0 10
1) "1912136916313##150481##1576140373##oLvqQwo25p5myELvO5VXj0k-7ngk##李伟测试##13926"
值为:
订单号##orderId##分配到期时间(uninx时间戳)##微信openid##设计师用户名##设计师userid
服务端窗口的变化
2019-12-14 01:29:15reactor-7 client-1 timer-2 taskid=0 :投递异步任务 data=1912136916313_150481_1576140373_oLvqQwo25p5myELvO5VXj0k-7ngk_李伟测试_13926
2019-12-14 01:29:15执行任务 id=0 data= 1912136916313_150481_1576140373_oLvqQwo25p5myELvO5VXj0k-7ngk_李伟测试_13926
任务完成 id=0 结果=1912136916313_150481_1576140373_oLvqQwo25p5myELvO5VXj0k-7ngk_李伟测试_13926已经派单过
task任务执行逻辑
数据分析
数据通过微信消息模板接口,将信息内容发送到客户微信公众号上
将监控日志和错误日志写入mysql数据库
三 项目代码结构
├── composer.json
├── composer.lock
├── swoole_server
│ ├── 2019-12-13-swoole.log
│ ├── 2019-12-14-swoole.log
│ ├── DispatchOrder.php
│ ├── pushClient.php
│ ├── pushServer.php
│ ├── Runtime.php
└── vendor
│ ├── 忽略php类库文件
│ ├── ....
conposer.json
{
"require": {
"predis/predis": "^1.1",
"catfan/medoo": "1.7.*"
}
}
pushServer.php
<?php
require __DIR__ . '/../vendor/autoload.php';
require './DispatchOrder.php';
class swoolePush
{
private $timerId;
private $timeCount = 0;
//微信消息模板id
private $templateid = "C3HSuUJdS86p_4gj4xxth943DdE2zkE3IxnlrK5MFTI";
public function receiveHandle($serv, $fd, $reactor_id, $data)
{
echo "reactor-{$reactor_id} Client-$fd 接受数据: data=$data " . PHP_EOL;
if ('start' == trim($data)) {//开启定时器
$redisServ = new Swoole\Coroutine\Redis;
$redisServ->connect('127.0.0.1', 6379);
//每隔5000ms触发一次
$this->timerId = Swoole\Timer::tick(5000, function ($timer_id) use ($redisServ, $serv, $reactor_id, $fd) {
$this->timeCount++;
echo date("Y-m-d H:i:s") . "reactor-{$reactor_id} client-{$fd} timer-{$timer_id}: 定时器第{$this->timeCount} 次执行" . PHP_EOL;
$item = $redisServ->lIndex("order_designer_list", -1);
if ($item) {
//订单号_orderId_分配到期时间(uninx时间戳)_openid_设计师用户名_设计师userid
$order = explode('##', $item);
if ($order[2] < time()) {
//投递异步任务
$task_id = $serv->task($item);
echo date("Y-m-d H:i:s") . "reactor-{$reactor_id} client-{$fd} timer-{$timer_id} taskid={$task_id} :投递异步任务 data=$item" . PHP_EOL;
$redisServ->rPop("order_designer_list");
}
}
});
} elseif ('stop' == trim($data)) {//清除定时器
echo "reactor-{$reactor_id} client-{$fd} timer-{$this->timerId} 定时器结束" . PHP_EOL;;
Swoole\Timer::clear($this->timerId);
}
}
/**
* 任务处理
* @param $serv
* @param $fd
*/
public function taskHandle($serv, $task_id, $from_id, $data)
{
echo date("Y-m-d H:i:s") . "执行任务 id=$task_id data= $data " . PHP_EOL;
//订单号_orderId_分配到期时间(uninx时间戳)_openid_设计师用户名_设计师userid
$order = explode('##', $data);
$returnMsg = "";
//派单逻辑
$dispatchOrderObj = new DispatchOrder();
if ($dispatchOrderObj->isOrderNeedGive($order[1])) {
//开始派单 orderGive($orderId, $designerUserId,$otherInfo=[])
$result = $dispatchOrderObj->orderGive($order[1], $order[5], [
'design_username' => $order[4], 'order_no' => $order[0], 'design_openid' => $order[3],
'design_templdateid' => "C3HSuUJdS86p_4gj4xxth943DdE2zkE3IxnlrK5MFTI"]);
if ($result['state'] != 1) {
//派单失败重新加入redis
$redisServ = new Swoole\Coroutine\Redis;
$redisServ->connect('127.0.0.1', 6379);
$redisServ->lPush("order_designer_list", $data);
$returnMsg = "{$order[0]}派单失败, errmsg:" . $result['message'];
} else {
$returnMsg = "{$order[0]}派单成功, errmsg:" . $result['message'];
}
} else {
$returnMsg = "{$order[0]}已经派单过";
}
//返回任务执行的结果
$serv->finish($returnMsg);
}
/**
* 任务完成通知
* @param $serv
* @param $fd
*/
public function finishHandle($serv, $task_id, $data)
{
echo "任务完成 id=$task_id 结果=$data" . PHP_EOL;
}
/**
* 连接开始
* @param $serv
* @param $fd
*/
public function connectHandle($serv, $fd)
{
echo "Client-$fd: 连接成功" . PHP_EOL;
}
/**
* 连接结束
* @param $serv
* @param $fd
*/
public function closeHandle($serv, $fd)
{
echo "Client-$fd: 连接结束" . PHP_EOL;
}
}
//创建Server对象，监听 127.0.0.1:9501端口
$serv = new Swoole\Server("127.0.0.1", 9501);
//设置异步任务的工作进程数量
$serv->set([
'task_worker_num' => 4 * 5,
'log_file' => date("Y-m-d") . '-swoole.log',
]);
$object = new swoolePush();
//监听连接进入事件
$serv->on('Connect', [$object, 'connectHandle']);
//监听数据接收事件
$serv->on('Receive', [$object, 'receiveHandle']);
//监听连接关闭事件
$serv->on('Close', [$object, 'closeHandle']);
//处理异步任务
$serv->on('task', [$object, 'taskHandle']);
//处理异步任务的结果
$serv->on('finish', [$object, 'finishHandle']);
//启动服务器
$serv->start();
pushClient.php
<?php
class pushClient
{
private $client;
public function __construct()
{
$this->client = new Swoole\Client(SWOOLE_SOCK_TCP);
if (!$this->client->connect('127.0.0.1', 9501, -1)) {
exit("connect failed. Error: {$this->client->errCode}" . PHP_EOL);
}
}
/**
* client发送开启定时器的指令
*/
public function startTimer()
{
$this->client->send("start");
}
/**
* client发送结束定时器的指令
*/
public function stopTimer()
{
$this->client->send("stop");
}
/**
* 自动销毁client
*/
public function __destruct()
{
$this->client->close();
}
}
//client连接开始
if (empty($argv[1])) {
exit("缺少参数, 参数:(string 'start|getinfo', int [num]) ");
}
$num = empty($argv[2]) ? 5 : intval($argv[2]);
//不同操作返回
//开启多个client连接,并发送start命令
if (trim($argv[1]) == 'start') {
for ($i = 0; $i < $num; $i++) {
$pushClient = new pushClient();
$pushClient->startTimer();
sleep(1);
}
} elseif (trim($argv[1]) == 'getinfo') {
//获取当前swoole的信息
$server = new Swoole\Server("127.0.0.1", 9501);
$arr = [
'client_connection_num' => count($server->connections)
];
var_dump($arr);
}
DispatchOrder.php
<?php
require __DIR__ . '/../vendor/autoload.php';
require './Runtime.php';
use Medoo\Medoo;
class DispatchOrder
{
private $configWchat = [
'app_id' => 'wx888888888888',
'secret' => '6d6e4f19888888888888888888',
];
private $configMysql = [
'database_type' => 'mysql',
'server' => '10.0.0.0',
'database_name' => 'db_test',
'username' => 'root',
'password' => '123456'
];
private $tokenWchat;
private $severMysql = null;
private $severRedis = null;
private $severRuntime = null;
private $result = ['state' => 0, 'message' => ''];
public function __construct()
{
//mysql
if ($this->severMysql == null) {
$this->severMysql = new Medoo($this->configMysql);
}
//日志
if ($this->severRuntime == null) {
$this->severRuntime = new Runtime(1, 1, 1, $this->severMysql);
}
}
/**
* 是否需要分配订单
* @param $orderId
* @return int 1需要分配 0不需要
*/
public function isOrderNeedGive($orderId)
{
$data = $this->severMysql->select('tg_order_task', [
'status',
], [
'order_id' => $orderId
]);
//1需要分配 0不需要
return (empty($data) || $data[0]['status'] == 1) ? 0 : 1;
}
/**
* @return Medoo
*/
public function orderGive($orderId, $designerUserId, $otherInfo = [])
{
//微信access_token
if ($this->severRedis == null) {
$redisServ = new Predis\Client([
'scheme' => 'tcp',
'host' => '127.0.0.1',
'port' => 6379]);
$this->severRedis = $redisServ;
}
$accessToken = $this->severRedis->get('swoole_order_dispatch_token');
$accessTokenArr = explode("##", $accessToken);
if (empty($accessToken) || $accessTokenArr[1] < time()) {
//设置token
$urlToken = "https://api.weixin.qq.com/cgi-bin/token?grant_type=client_credential&app . $this->configWchat['secret'];
$res = $this->httpRequest($urlToken, "get");
$arr = json_decode($res, true);
if (empty($arr['access_token'])) {
$this->severRuntime->log("请求微信token接口失败,返回:" . $res);
}
$expiresTime = (string)(time() + 7200);
$value = (string)($arr['access_token'] . "##" . $expiresTime);
$this->severRedis->set('swoole_order_dispatch_token', $value);
$this->tokenWchat = $arr['access_token'];
} else {
$this->tokenWchat = $accessTokenArr[0];
}
// var_dump($this->tokenWchat);
//1.更新订单状态
$updateData = [
'design_userid' => $designerUserId,
'design_username' => $otherInfo['design_username'],
'design_time' => date("Y-m-d H:i:s"),
'status' => 1,//0 待抢单 1待设计 2已设计 3已完成4已取消5驳回
];
$res = $this->severMysql->update('tg_order_task', [
'status' => 1,
], [
'order_id' => $orderId
]);
// var_dump($res);
if (0 == $res->rowCount()) {
$this->result['message'] = "更新任务订单失败,订单号:" . $otherInfo['order_no'];
$this->severRuntime->log($this->result['message']);
return $this->result;
}
//2.微信模板信息推送
$data = [
'touser' => $otherInfo['design_openid'],
'template_id' => $otherInfo['design_templdateid'],
'url' => '',
'data' => [
'keyword1' => ['value' => $otherInfo['order_no']],
'keyword2' => ['value' => date("Y-m-d H:i:s")]
],
];
$url = 'https://api.weixin.qq.com/cgi-bin/message/template/send?access_token=' . $this->tokenWchat;
$res = $this->httpRequest($url, "post", json_encode($data));
// var_dump($res);
$resObj = json_decode($res);
if (0 != $resObj->errcode) {
$this->result['message'] = "{$otherInfo['order_no']}微信推送失败,$res";
$this->severRuntime->log($this->result['message']);
return $this->result;
}
$this->result['state'] = 1;
$this->result['message'] = "执行成功,订单号:" . $otherInfo['order_no'];
return $this->result;
}
/**
* CURL请求
*
* @param string 请求url地址
* @param method 请求方法 get post
* @param null $postfields post数据数组
* @param array $headers 请求header信息
* @param bool|false $debug 调试开启 默认false
* @return mixed
*/
private function httpRequest($url, $method, $postfields = null, $headers = [], $debug = false)
{
$method = strtoupper($method);
$ci = curl_init();
/* Curl settings */
curl_setopt($ci, CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_1_0);
curl_setopt($ci, CURLOPT_USERAGENT, "Mozilla/5.0 (Windows NT 6.2; WOW64; rv:34.0) Gecko/20100101 Firefox/34.0");
curl_setopt($ci, CURLOPT_CONNECTTIMEOUT, 60); /* 在发起连接前等待的时间，如果设置为0，则无限等待 */
curl_setopt($ci, CURLOPT_TIMEOUT, 7); /* 设置cURL允许执行的最长秒数 */
curl_setopt($ci, CURLOPT_RETURNTRANSFER, true);
switch ($method) {
case "POST":
curl_setopt($ci, CURLOPT_POST, true);
if (!empty($postfields)) {
$tmpdatastr = is_array($postfields) ? http_build_query($postfields) : $postfields;
curl_setopt($ci, CURLOPT_POSTFIELDS, $tmpdatastr);
}
break;
default:
curl_setopt($ci, CURLOPT_CUSTOMREQUEST, $method); /* //设置请求方式 */
break;
}
$ssl = preg_match('/^https:\/\//i', $url) ? TRUE : FALSE;
curl_setopt($ci, CURLOPT_URL, $url);
if ($ssl) {
curl_setopt($ci, CURLOPT_SSL_VERIFYPEER, FALSE); // https请求 不验证证书和hosts
curl_setopt($ci, CURLOPT_SSL_VERIFYHOST, FALSE); // 不从证书中检查SSL加密算法是否存在
}
curl_setopt($ci, CURLOPT_MAXREDIRS, 2); /* 指定最多的HTTP重定向的数量，这个选项是和CURLOPT_FOLLOWLOCATION一起使用的 */
curl_setopt($ci, CURLOPT_HTTPHEADER, $headers);
curl_setopt($ci, CURLINFO_HEADER_OUT, true);
$response = curl_exec($ci);
$requestinfo = curl_getinfo($ci);
$http_code = curl_getinfo($ci, CURLINFO_HTTP_CODE);
if ($debug) {
echo "=====post data======\r\n";
var_dump($postfields);
echo "=====info===== \r\n";
print_r($requestinfo);
echo "=====response=====\r\n";
print_r($response);
}
curl_close($ci);
return $response;
}
/**
* 关闭资源链接
*/
public function __destruct()
{
}
}
Runtime.php
<?php
require __DIR__ . '/../vendor/autoload.php';
use Medoo\Medoo;
class Runtime
{
//isMysql是否写入数据 isFile是否写入文件 isConsole是否console输入
private $isMysql;
private $isFile;
private $isConsole;
private $startTime = 0;
private $stopTime = 0;
private $severMysql = null;
public function __construct($isMysql, $isFile, $isConsole, $severMysql)
{
$this->isMysql = $isMysql;
$this->isFile = $isFile;
$this->isConsole = $isConsole;
//mysql
if ($this->severMysql == null) {
$this->severMysql = $severMysql;
}
}
/**
* 日志记录
* @param $content
* @param $runtime
*/
public function log($content, $runtime = 0, $mode = [])
{
//数据库
if ($this->isMysql) {
$data = [
'content' => $content,
'runtime' => $runtime,
'w_time' => date("Y-m-d H:i:s"),
];
$this->severMysql->insert('tg_log_order_dispatch', $data);
}
//写入文件
if ($this->isFile) {
file_put_contents(date("Y-m-d").'-runtime.log', $content."\n", FILE_APPEND);
}
//命令窗口
if ($this->isConsole) {
echo date("Y-m-d H:i:s") . " " . $content . PHP_EOL;
}
}
//开始运行时间
public function start()
{
$this->startTime = $this->getMicrotime();
}
//结束时间
public function stop()
{
$this->stopTime = $this->getMicrotime();
}
//开始和结束之间总时长
public function spent()
{
return ($this->stopTime - $this->startTime);
}
private function getMicrotime()
{
list($usec, $sec) = explode(' ', microtime());
return ((float)$usec + (float)$sec);
}
}
