原生workman实现及时消息推送

需求：

共享单车关闭后，会通过设置的url，进行回调，发送数据。那么相对于共享单车关锁来说，回调给url的数据，对于客户端来说是异步操作，那么如何使回调变成同步操作，客户端能立即收到关锁之后的信息呢？

问题：

由于进程之间申请的内存不能共享，所以当你直接调用workman里面封装的方法时，会无法拿到数据。所以好多人就会出现，在回调里面调用推送的方法，提示 未定义方法或方法不存在的情况

workman 代码:

use Workerman\Worker;

use Workerman\Lib\Timer;

//引入workman自动加载类

require_once __DIR__.'/../../Workerman/Autoloader.php';

//设置心跳时间 55秒

define('Heartbeat',55);

$ws = new Worker('websocket://0.0.0.0:8013');

// 启动1个进程对外提供服务

$ws->count = 1;

// 新增加一个属性，用来保存uid到connection的映射(uid是用户id或者客户端唯一标识)

$ws->uidConnections = array();

$ws->onWorkerStart = function ($ws){

//设置一个 间隔 1秒的定时

Timer::add(1,function ()use($ws){

$time_now = time();

foreach($ws->connections as $connection) {

// 有可能该connection还没收到过消息，则lastMessageTime设置为当前时间

if (empty($connection->lastMessageTime)) {

$connection->lastMessageTime = $time_now;

continue;

}

// 上次通讯时间间隔大于心跳间隔，则认为客户端已经下线，关闭连接

if ($time_now - $connection->lastMessageTime > Heartbeat) {

$connection->close();

}

}

});

// 开启一个内部端口，方便内部系统推送数据，Text协议格式 文本+换行符

$inner_text_worker = new Worker('text://0.0.0.0:5678');

$inner_text_worker->onMessage = function($connection, $buffer)

{

// $data数组格式，里面有uid，表示向那个uid的页面推送数据

$data = json_decode($buffer, true);

$uid = $data['uid'];

// 通过workerman，向uid的页面推送数据

$ret = sendMessageByUid($uid, $buffer);

// 返回推送结果

$connection->send($ret ? 'ok' : 'fail');

};

// ## 执行监听 ##

$inner_text_worker->listen();

};

// 当收到客户端发来的数据后返回hello $data给客户端

$ws->onMessage = function($connection, $data)

{

global $ws;

// 给connection临时设置一个lastMessageTime属性，用来记录上次收到消息的时间

$connection->lastMessageTime = time();

if(!isset($connection->uid))

{

//简单的验证

if(intval($data))

{

// 没验证的话把第一个包当做uid

$connection->uid = $data;

/* 保存uid到connection的映射，这样可以方便的通过uid查找connection，

* 实现针对特定uid推送数据

*/

echo $connection->uid."已上线\n";

$ws->uidConnections[$connection->uid] = $connection;

}

return ;

}

};

// 当有客户端连接断开时

$ws->onClose = function($connection)

{

global $ws;

if(isset($connection->uid))

{

echo $connection->uid."已断开\n";

// 连接断开时删除映射

unset($ws->uidConnections[$connection->uid]);

}

};

// 针对uid推送数据

function sendMessageByUid($uid, $message)

{

global $ws;

if(isset($ws->uidConnections[$uid]))

{

$connection = $ws->uidConnections[$uid];

$connection->send($message);

return true;

}

return false;

}

// 运行worker

Worker::runAll();

callback 代码：

// 建立socket连接到内部推送端口

$client = stream_socket_client('tcp://127.0.0.1:5678', $errno, $errmsg, 3);

// 推送的数据，包含uid字段，表示是给这个uid推送

// 发送数据，注意5678端口是Text协议的端口，Text协议需要在数据末尾加上换行符

fwrite($client, json_encode($row)."\n");

// 读取推送结果,并写入日志

Log::notice('订单消息推送'.fread($client, 8192));

fclose($client);

注意：

进程之间内存不能共享，所以在websocket Worker内部建立一个text Worker，这两个Worker是同一个进程，可以方便的共享客户端连接

设置了全局变量，全局变量用完之后要销毁删除，否则造成 内存泄漏 导致程序boom

设置websocket心跳，保证长连接在线

客户端收到推送的订单后，进行一系列操作