onWorkerStart = function($ws_worker) { // 将db实例存储在全局变量中(也可以存储在某类的静态成员中) global $websocketDB; global $mqtt; $websocketDB = new \Workerman\MySQL\Connection('rm-wz98r5cn33zq4ou980o.mysql.rds.aliyuncs.com', '3306', 'idcol20', 'idcol@1234', 'idcol'); global $devList; $devList = array(); global $devUpdateList; $devUpdateList = array(); $options = array( 'username'=>'weclouds', 'password'=>'weclouds@1234' ); $mqtt = new \Workerman\Mqtt\Client('mqtt://streetserver.weclouds.xyz:1883',$options); $mqtt->onConnect = function($mqtt) { $room = array( '/IDCOL/CmdInput/#' => 2, '/IDCOL/CmdOutput/#' => 2, ); $mqtt->subscribe($room, null, function($exception, $granted){}); $mqtt->onMessage = function($room, $message){ global $devUpdateList; global $StrategyReport; $idArr = explode('/', $room); $id = $idArr[count($idArr) - 1]; $id = strtolower($id); if (strpos($room,'CmdOutput')) { $msg = unpack('H*',$message); if (isset($devUpdateList[$id]) && !empty($devUpdateList[$id])) { $temp = $devUpdateList[$id]; $type = substr($msg[1], 0,2); if ($type == '99') { $status = substr($msg[1],18,2); $devUpdateList[$id]['status'] = $status; if ($status == '00') { $devUpdateList[$id]['type'] = 1; }elseif ($status == '01') { $devUpdateList[$id]['type'] = 2; } } } } }; }; $mqtt->connect(); Timer::add(0.5, function()use($ws_worker){ // global $clientArr; global $websocketDB; $connections = $ws_worker->connections; // var_dump($connections); $data = $websocketDB->select('id,client,msg')->from('message')->query(); if (!empty($data)) { foreach ($data as $v) { if (!empty($connections[$v['client']])) $connections[$v['client']]->send($v['msg']); $websocketDB->query("DELETE FROM `message` WHERE id='{$v['id']}'"); } } }); // 下发设备升级指令 Timer::add(1, function()use($ws_worker){ global $mqtt; global $devUpdateList; global $devList; global $websocketDB; $temp = $devList; if (!empty($temp)) { foreach ($temp as $v) { $lamp = $websocketDB->query("select manu,address,id from lampinfo where id = ".$v['lamp_id']); if (empty($lamp) || empty($lamp[0]['id'])) { # code... } } } }); // 设备升级进度 Timer::add(2, function()use($ws_worker){ // global $clientArr; // global $websocketDB; // $connections = $ws_worker->connections; // // var_dump($connections); // $data = $websocketDB->select('id,client,msg')->from('message')->query(); // if (!empty($data)) { // foreach ($data as $v) { // if (!empty($connections[$v['client']])) $connections[$v['client']]->send($v['msg']); // $websocketDB->query("DELETE FROM `message` WHERE id='{$v['id']}'"); // } // } }); }; // 启动1个进程对外提供服务 $ws_worker->count = 1; // 客户端连接回调 $ws_worker->onConnect = function($connection) { // global $clientArr; // 标记连接的客户端 // $key=uniqid(); $key = $connection->id; $data = array('code'=>'0000','msg'=>'操作成功','data'=>['clientid'=>$key,'status'=>1]); $connection->send(json_encode($data)); // $clientArr[$key] = $connection; }; $ws_worker->onMessage = function($connection, $data) { global $devList; $res = json_decode($data,true); if ($res && isset($res['type'])) { if ($res['type'] == 'updateDev' && isset($res['id']) && !empty($res['id'])){ $arr = explode(','$res['id']); if (!empty($arr)) { foreach ($arr as $key) { if (is_int($key) && !empty($key)) { $devList[] = array('lamp_id'=>intval($key),'client'=>$connection); } } }else{ $connection->close(); } }else{ $connection->close(); } }else{ $connection->close(); } }; // 客户端断开连接 // $ws_worker->onClose = function($connection) // { // global $clientArr; // if (!empty($clientArr)) { // foreach ($clientArr as $k => $v) { // if ($v == $connection) { // unset($clientArr[$k]); // break; // } // } // } // }; // 运行worker Worker::runAll();