发表新帖

请问关于定时器超时或者协程超时被关闭无法退出问题

### 问题描述
各位朋友 请教一下我的问题， 项目需求是硬件设备Tcp client长连接服务器，控制端为http client,
当调用方发起http控制转发tcp控制指令，正常情况设备完成指令回复服务器，服务器返回http客户结果，
如设备异常没有回复服务器，那么超过5秒后，服务器回复http客户端失败

我用的是swoole多端口服务，http监听9501，tcp监听9502 用内存表记录设备client FD
一个控制的流程为
1. 当调用方发起http控制请求
onRequest事件，我用了两个方案1.一次性定时器 twosee大神提示我改用协程方式
----定时器方案
1. 根据对应关系找到设备FD,发出tcp控制指令
2. response的FD保存在内存表
3. 启用一个一次性定时器 参数为response的FD值，超时5秒后，内存表里检查 response FD是否存在
   如果存在 Swoole\Http\Response::create出response对象，返回失败，清除内存表response FD记录项
4. response->detach() 退出  onRequest事件 
   
----协程方案
1. 根据对应关系找到设备FD,发出tcp控制指令
2. response的FD保存在内存表
3. 启用一个swoole协程函数 参数为response的FD值，超时5秒后，内存表里检查 response FD是否存在
   如果存在 Swoole\Http\Response::create出response对象，返回失败，清除内存表response FD记录项
4. response->detach() 退出  onRequest事件 
   
   
正常情况设备完成指令回复服务器，服务器OnReceive事件中投递到 task中执行以下步骤
1. 取出 内存表里 response FD
2. Swoole\Http\Response::create出response对象，返回成功 清除内存表response FD记录项

目前的情况是，http转发指令正常，就是设备没有回复的情况下，偶尔能正常工作，多数，超时不能返回，服务端报以下错误，一直把http hold住
终止后台程序后才释放，
 
	WARNING	swWorker_reactor_is_empty (ERRNO 9012): worker exit timeout, forced to terminate

### Swoole版本，PHP版本，以及操作系统版本信息
swoole 4.4.16  PHP 7.2.18 (cli)  CentOS Linux release 7.5.1804 (Core)

### 相关代码

public function __construct() {
        //$this->serv = new swoole_websocket_server("0.0.0.0", 9501);
        $this->serv = new Server("0.0.0.0", 9501);
        $this->serv->set([
            'worker_num'      => 2, //开启2个worker进程
            'max_request'     => 4, //每个worker进程 max_request设置为4次
            'task_worker_num' => 4, //开启4个task进程
            'dispatch_mode'   => 4, //数据包分发策略 - IP分配
            'daemonize'       => false, //守护进程(true/false)
            'enable_coroutine' => true,

]);

$this->serv->on('Start', [$this, 'onStart']);
        $this->serv->on('Open', [$this, 'onOpen']);
        $this->serv->on("Message", [$this, 'onMessage']);
        $this->serv->on("Request", [$this, 'onRequest']);
        $this->serv->on("Close", [$this, 'onClose']);
        $this->serv->on("Task", [$this, 'onTask']);
        $this->serv->on("Finish", [$this, 'onFinish']);

//监听 9502 端口
        $tcp = $this->serv->listen("0.0.0.0", 9502, SWOOLE_SOCK_TCP);
        $tcp->set([
            'worker_num'      => 2, //开启2个worker进程
            'max_request'     => 4, //每个worker进程 max_request设置为4次
            'dispatch_mode'   => 2, //数据包分发策略 - 固定模式

]);
        $tcp->on("connect", [$this, 'onTcpConnect']);
        $tcp->on("Receive", [$this, 'onReceive']);

//监听 9503 端口 UDP
        $udp = $this->serv->listen("0.0.0.0", 9503, SWOOLE_SOCK_UDP);
        $udp->set([
            'worker_num'      => 2, //开启2个worker进程
            'max_request'     => 4, //每个worker进程 max_request设置为4次
            'dispatch_mode'   => 2, //数据包分发策略 - 固定模式
        ]);
        $udp->on("Packet", [$this, 'onPacket']);

$this->swtbl = new SWtable();
        $this->serv->start();
    }

public function onStart($serv) {
        echo "-------- onStart SWOOLE ".SWOOLE_VERSION . " 服务已启动 ---------".PHP_EOL;
        echo "master_pid: {$serv->master_pid} manager_pid:".$serv->manager_pid.PHP_EOL;
    }

public function onOpen($serv, $request) {
        echo "#### onOpen ####".PHP_EOL;
        echo "server: handshake success with fd{$request->fd}".PHP_EOL;
        $serv->task([
            'type' => 'login'
        ]);
        echo "########".PHP_EOL.PHP_EOL;
    }

public function onTask($serv, $task_id, $from_id, $data) {
        echo "#### onTask ####".PHP_EOL;
        echo "#{$serv->worker_id} onTask: [PID={$serv->worker_pid}]: task_id={$task_id}".PHP_EOL;
        $msg = '';
        switch ($data['type']) {
            case 'login':
                $msg = '我来了...';
                break;

case 'speak':
                $msg = $data['msg'];
                break;

case 'tcpRcv':
                $this->tcpRecvPro($data['srcFD'],$data['data']);
                break;
        }
        /*
        foreach ($serv->connections as $fd) {
            $connectionInfo = $serv->connection_info($fd);
            if (isset($connectionInfo['websocket_status']) && $connectionInfo['websocket_status'] == 3) {
                $serv->push($fd, $msg); //长度最大不得超过2M
            }
        }
        */
        $serv->finish($data);
    }
    /*
      call from task
      onReceive -> task -> tcprecvpro
     */
    public function tcpRecvPro($srcFD,$dataRec){
        $ilen0 = strlen($dataRec);

$szdevtype = "0";
        $szdevsn = "";
        $ilen = 0;
        $aryd = unpack("C*",$dataRec);
        $ilen0 = count($aryd);

if ($ilen0 > 23){ // get dev sn yinze
             if ($aryd[6] == 0x43 && $aryd[7] == 0x55){
                  for ($ix = 0;$ix < 16;$ix++){
                         $szdevsn = $szdevsn  . chr($aryd[$ix + 6]);
                  }
                  if (strlen($szdevsn) == 16) {
                        if (substr($szdevsn,0,5) == "CU-BM") {
                              $szdevtype= "A";
                        } else  if (substr($szdevsn,0,5) == "FC-*(") {
                              $szdevtype= "C";
                        }
                  }
                  echo "get dev sn:".$szdevsn.PHP_EOL;
                  if ($aryd[25] == 0x19 && $aryd[26] == 0x22){

}
             }
        }
        if ($szdevtype != "0"){// 被识别类型的设备
              if (strlen($szdevsn) > 10 ){// get devsn 有效的devSN
                   $FDObj = $this->swtbl->getI(0,$srcFD);
                   if (!$FDObj){
                        $aryv = array("devSN" => $szdevsn,'uptime' => time());
                        $res1 = $this->swtbl->updateItem(0,$srcFD,$aryv);
                        if (!$res1){
                              echo "保存错误....".PHP_EOL;
                        }
                        $aryv = array("swFD" => $srcFD);
                        $this->swtbl->updateItem(3,$szdevsn,$aryv);
                   } else {
                       $szt1 = $FDObj["devSN"];
                       if (strlen($szt1) > 10){
                            $devOBJ = $this->swtbl->getI(3,$szt1);
                            if ($devOBJ){
                                 if ($devOBJ["swFD"] != $srcFD){
                                     $aryv = array("swFD" => $srcFD);
                                     $this->swtbl->updateItem(3,$szdevsn,$aryv);
                                 }
                            } else {
                                $aryv = array("swFD" => $srcFD,"devType" =>$szdevtype);
                                $this->swtbl->updateItem(3,$szdevsn,$aryv);
                            }
                       }
                   }
              }
        }

$msg1 = bytestr2string($dataRec,$ilen);
        echo "len:$ilen Msg:".$msg1."  FD:".$srcFD.PHP_EOL;
        return;
        $tcpdev = $this->swtbl->getI(1, $srcFD);
        if ($tcpdev){
               echo "find dev FD_httprecent is ". $tcpdev['FD_httprecent'].PHP_EOL;
               if ($tcpdev['FD_httprecent']) {
                       $resp = \Swoole\Http\Response::create($tcpdev['FD_httprecent']);
                       $resp->end("<h1>tcp recv to http  ".date("Y-m-d H:i:s",time())."</h1>");
                       $this->swtbl->delI(4,$tcpdev['FD_httprecent']); // 用完即弃
                       // 更新　tableFDSub　FD_httprecent　置　0
                       //$this->swtbl->updateTcp2HttpFD($srcFD,0);
                       $aryt = array("FD_httprecent" =>0);
                       $this->swtbl->updateItem(1,$srcFD,$aryt);
               }
        } else{
            echo "not found: srcFD:".$srcFD.PHP_EOL;
        }
    }

public function onFinish($serv,$task_id, $data) {
        echo "Task onFinish  {$task_id} 已完成".PHP_EOL;
    }

public function onClose($serv, $fd) {
        echo "#### onClose ####".PHP_EOL;
        echo "client {$fd} closed".PHP_EOL;
        echo "########".PHP_EOL.PHP_EOL;
    }

public function onMessage($serv, $frame) {
        echo "#### onMessage ####".PHP_EOL;
        echo "receive from fd{$frame->fd}:{$frame->data},opcode:{$frame->opcode},fin:{$frame->finish}".PHP_EOL;
        $serv->task(['type' => 'speak', 'msg' => $frame->data]);
        echo "########".PHP_EOL.PHP_EOL;
    }

public function onRequest($request, $response) {
        echo "-- onRequest response->fd: ".$response->fd.PHP_EOL;
        $response->header("Content-Type", "text/html; charset=utf-8");
        $server = $request->server;
        $path_info    = $server['path_info'];
        $request_uri  = $server['request_uri'];

if ($path_info == '/favicon.ico' || $request_uri == '/favicon.ico') {
            return $response->end();
        }

echo "PATH_INFO:".$path_info. "  request_uri:" .$request_uri . PHP_EOL;
        $addinfo = "";

if ($request_uri == '/') {
              $addinfo = "<a href='/sys'>系统</a>";
        } else if ($request_uri == '/pro') {

echo "got pro dir ... des FD" . $request->get['afd'] . PHP_EOL;

$tcpFD = (int)$request->get['afd'];
            $httpFD = (int)$response->fd;
            $this->senddir($tcpFD,0);

$curtime = time();
            $aryHttp = array('devFD' =>$tcpFD ,'cmdTime' => $curtime,'expTime' =>  $curtime  + 5000 );
            $this->swtbl->updateItem(4,$response->fd,$aryHttp);

$aryFD = array('FD_httprecent' =>$httpFD ,'cmdTime' => $curtime);
            $this->swtbl->updateItem(1, $tcpFD,$aryFD);

//$this->swtbl->updateTcp2HttpFD($request->get['afd'],(int)$response->fd);
            // http fd => fd
            //$this->swtbl->updateHttptable($response->fd,$request->get['afd']);
            // fdsub fd => httpfd
            $sttime = date("Y-m-d H:i:s",time());
            //超时定时器
            /*
            $timer = \Swoole\Timer::after(3000, function() use ($tcpFD,$httpFD,$sttime) {
                $endtime = date("Y-m-d H:i:s",time());
                $arett = $this->swtbl->geti(4,$httpFD);

if (!$arett){
                    echo "not found  http fd----------------".$httpFD.PHP_EOL;
                      return;
                }
                $this->swtbl->delI(4,$httpFD);
                $resp = \Swoole\Http\Response::create($httpFD);
                $resp->end("<h1>http call expire ".$sttime." ---<br>". $endtime." </h1>");
            });
           */

}
        $html = "<h1>你好 Swoole.</h1>" . $addinfo;
        $response->end($html);
    }

public function httpExpro($HttpFD){
        /*
        $sttime = date("Y-m-d H:i:s",time());
        $timer = \Swoole\Timer::after(3000, function() use ($HttpFD,$sttime) {
            $endtime = date("Y-m-d H:i:s",time());
            $arett = $this->swtbl->geti(4,$HttpFD);

if (!$arett){
                echo "not found  http fd----------------".$HttpFD.PHP_EOL;
                return;
            }
            $this->swtbl->delI(4,$HttpFD);
            $resp = \Swoole\Http\Response::create($HttpFD);
            $resp->end("<h1>http call expire ".$sttime." ---<br>". $endtime." </h1>");
        });
        */

\go(function () use ($HttpFD) {//use 关键字 函数闭包 调用外部变量
            $sttime = date("Y-m-d H:i:s",time());
            echo "enter exp proc ". $sttime.PHP_EOL;
            \Co::sleep(5);
            //\Swoole\Coroutine::sleep(5);
            $endtime = date("Y-m-d H:i:s",time());
            echo "aft sleep 5 ".$endtime.PHP_EOL;
            $arett = $this->swtbl->geti(4,$HttpFD); //从内存表取出 resp fd()
            if (!$arett){
                echo "not found  http fd----------------".$HttpFD.PHP_EOL;
                return;
            }
            $this->swtbl->delI(4,$HttpFD);
            $resp = \Swoole\Http\Response::create($HttpFD);
            $resp->end("<h1>http call expire $sttime<br>$endtime</h1>");
        });
    }

public function onReceive($serv, $fd, $from_id, $data) {
        echo "#### onReceive #### fd ". $fd .PHP_EOL;

$serv->task([
            'type' => 'tcpRcv',
            'data' =>  $data,
            'srcFD' => $fd
        ]);
    }

/*
          Tcp connected
    */
    public function onTcpConnect($srv, $fd) {
        var_dump("Tcp connected: ".$fd);
       // var_dump($srv->connection_info($fd));
    }

public function onPacket($serv, $data, $clientInfo) {
        echo "#### onPacket ####".PHP_EOL;
        $serv->sendto($clientInfo['address'], $clientInfo['port'], "Server ".$data);
        //var_dump($clientInfo);
        echo "data is $data".PHP_EOL;
        if (strlen($data) > 3){
             $szaryt = explode(",",$data);
             if (count($szaryt) > 3) {
               /*
                   5,5,0,5
                        5 = 指令   (5 表顺序ID)  0 read 1 键值
                   5,5,1,5,httpFD|3
                        5 = 指令   (5 表顺序ID)  (1 write) 5 key (httpFD键值 3内容)
                   set get table value [1] table seqe  [2] type 0 get 1 set  [3] key [4] value(键值对 | 分隔)
                 */
                 if ($szaryt[0] == 5){
                       $itabid = (int)$szaryt[1];
                       if ($szaryt[2] == "0") {
                            $retv = $this->swtbl->getI($itabid, $szaryt[3]);
                            var_dump($retv);
                            echo PHP_EOL;
                       } else  if ($szaryt[2] == "1"){
                           $aryv = explode("|",$szaryt[4]); // 要设置的 字段名 内容键值对
                           $v1 = (int) $aryv[1];
                           $parakv = array( $aryv[0] => $v1);
                           //$parakv = $aryv[0].",".$aryv[1];

$bret = $this->swtbl->updateItem($itabid, $szaryt[3],$parakv );
                           if ($bret){
                                 echo "write table ok".PHP_EOL;
                           } else {
                                 echo "write table failed".PHP_EOL;
                           }
                       }
                 }
             }
        } else if ($data == "1"){
             echo "desFDInv ".$this->desFDInv.PHP_EOL;
        } else if ($data == "2"){
            $this->desFDInv1 = 6;
        } else if ($data == "3"){
            echo "desFDInv ".$this->desFDInv1.PHP_EOL;
        }

}
    //--------

public function senddir($desFD,$idir){
        //开1 2 号门 "7E43552D424D3630323230313234303030FFFFFFFF2F6DE76B03030000000004010100006B7E"
        //CU-BM60220124000
        //FC-8940A47010002
        //    7E46432D38393430413437303130303032FFFFFFFF6FF40B560303000000000401010000267E
        $arydir = str2Byt("7E46432D38393430413437303130303032FFFFFFFF6FF40B560303000000000401010000267E");
        $szdir = bye2str($arydir);

if ($idir == 0){// opendoor
            $this->serv->send($desFD, $szdir);
            echo "sent...".PHP_EOL;
        }
    }
}

发布于2年前 · 14 次浏览 · 来自提问

ufuture