GatewayWorker的onWebSocketConnect回调函数中能否使用while(true)循环来轮询外部HTTP接口？

prcvoldermort 2022-01-30

各位大大，新年快乐！
最近在开发一个websocket服务，需要根据websocket连接上来时携带的参数来创建一个个的session，分别轮询外部的接口服务器，来获取实时的话务系统状态与话务事件。于是我考虑在GatewayWorker 的onWebSocketConnect回调函数中使用while(true)循环来不断的轮询接口，直到这个session连接断开。
HTTP的请求基于GuzzleHttp开发，在实际运行中，轮询个5到6次后会报process_timeout的错误，堆栈信息如下：

< HTTP/1.1 200 
< Transfer-Encoding: chunked
< Date: Sun, 30 Jan 2022 01:41:50 GMT
< 
* Connection #0 to host 192.168.15.4 left intact
* Hostname 192.168.15.4 was found in DNS cache
*   Trying 192.168.15.4...
* TCP_NODELAY set
* Connected to 192.168.15.4 (192.168.15.4) port 8080 (#0)
> POST /AeonixProxy/proxy HTTP/1.1
Host: 192.168.15.4:8080
User-Agent: GuzzleHttp/7
SOAPAction: getEventsRequest
Content-Type: text/xml; charset=UTF-8
Content-Type: text/xml; charset=UTF-8
Content-Length: 294

* upload completely sent off: 294 out of 294 bytes
process_timeout:
#1 /home/vagrant/code/AeonixWebsocketService/vendor/guzzlehttp/guzzle/src/Handler/CurlMultiHandler.php(183): GuzzleHttp\Handler\CurlMultiHandler->tick()
#2 /home/vagrant/code/AeonixWebsocketService/vendor/guzzlehttp/promises/src/Promise.php(248): GuzzleHttp\Handler\CurlMultiHandler->execute()
#3 /home/vagrant/code/AeonixWebsocketService/vendor/guzzlehttp/promises/src/Promise.php(224): GuzzleHttp\Promise\Promise->invokeWaitFn()
#4 /home/vagrant/code/AeonixWebsocketService/vendor/guzzlehttp/promises/src/Promise.php(269): GuzzleHttp\Promise\Promise->waitIfPending()
#5 /home/vagrant/code/AeonixWebsocketService/vendor/guzzlehttp/promises/src/Promise.php(226): GuzzleHttp\Promise\Promise->invokeWaitList()
#6 /home/vagrant/code/AeonixWebsocketService/vendor/guzzlehttp/promises/src/Promise.php(62): GuzzleHttp\Promise\Promise->waitIfPending()
#7 /home/vagrant/code/AeonixWebsocketService/app/Aeonix/Telephony.php(311): GuzzleHttp\Promise\Promise->wait()#8 /home/vagrant/code/AeonixWebsocketService/app/GatewayWorker/Events.php(149): App\Aeonix\Telephony->sendToProxyEndpoint()
#9 [internal function]: App\GatewayWorker\Events::onWebSocketConnect()
#10 /home/vagrant/code/AeonixWebsocketService/vendor/workerman/gateway-worker/src/BusinessWorker.php(419): call_user_func()
#11 /home/vagrant/code/AeonixWebsocketService/vendor/workerman/workerman/Connection/TcpConnection.php(656): GatewayWorker\BusinessWorker->onGatewayMessage()
#12 /home/vagrant/code/AeonixWebsocketService/vendor/workerman/workerman/Events/Select.php(292): Workerman\Connection\TcpConnection->baseRead()
#13 /home/vagrant/code/AeonixWebsocketService/vendor/workerman/workerman/Worker.php(2408): Workerman\Events\Select->loop()
#14 /home/vagrant/code/AeonixWebsocketService/vendor/workerman/gateway-worker/src/BusinessWorker.php(197): Workerman\Worker->run()
#15 /home/vagrant/code/AeonixWebsocketService/vendor/workerman/workerman/Worker.php(1541): GatewayWorker\BusinessWorker->run()
#16 /home/vagrant/code/AeonixWebsocketService/vendor/workerman/workerman/Worker.php(1371): Workerman\Worker::forkOneWorkerForLinux()
#17 /home/vagrant/code/AeonixWebsocketService/vendor/workerman/workerman/Worker.php(1345): Workerman\Worker::forkWorkersForLinux()
#18 /home/vagrant/code/AeonixWebsocketService/vendor/workerman/workerman/Worker.php(546): Workerman\Worker::forkWorkers()
#19 /home/vagrant/code/AeonixWebsocketService/app/Console/Commands/GatewayWorkerServer.php(63): Workerman\Worker::runAll()
#20 /home/vagrant/code/AeonixWebsocketService/app/Console/Commands/GatewayWorkerServer.php(55): App\Console\Commands\GatewayWorkerServer->start()
#21 /home/vagrant/code/AeonixWebsocketService/vendor/laravel/framework/src/Illuminate/Container/BoundMethod.php(36): App\Console\Commands\GatewayWorkerServer->handle()
#22 /home/vagrant/code/AeonixWebsocketService/vendor/laravel/framework/src/Illuminate/Container/Util.php(40): Illuminate\Container\BoundMethod::Illuminate\Container\{closure}()
#23 /home/vagrant/code/AeonixWebsocketService/vendor/laravel/framework/src/Illuminate/Container/BoundMethod.php(93): Illuminate\Container\Util::unwrapIfClosure()
#24 /home/vagrant/code/AeonixWebsocketService/vendor/laravel/framework/src/Illuminate/Container/BoundMethod.php(37): Illuminate\Container\BoundMethod::callBoundMethod()
#25 /home/vagrant/code/AeonixWebsocketService/vendor/laravel/framework/src/Illuminate/Container/Container.php(653): Illuminate\Container\BoundMethod::call()
#26 /home/vagrant/code/AeonixWebsocketService/vendor/laravel/framework/src/Illuminate/Console/Command.php(136): Illuminate\Container\Container->call()
#27 /home/vagrant/code/AeonixWebsocketService/vendor/symfony/console/Command/Command.php(298): Illuminate\Console\Command->execute()
#28 /home/vagrant/code/AeonixWebsocketService/vendor/laravel/framework/src/Illuminate/Console/Command.php(121): Symfony\Component\Console\Command\Command->run()
#29 /home/vagrant/code/AeonixWebsocketService/vendor/symfony/console/Application.php(1005): Illuminate\Console\Command->run()
#30 /home/vagrant/code/AeonixWebsocketService/vendor/symfony/console/Application.php(299): Symfony\Component\Console\Application->doRunCommand()
#31 /home/vagrant/code/AeonixWebsocketService/vendor/symfony/console/Application.php(171): Symfony\Component\Console\Application->doRun()
#32 /home/vagrant/code/AeonixWebsocketService/vendor/laravel/framework/src/Illuminate/Console/Application.php(94): Symfony\Component\Console\Application->run()
#33 /home/vagrant/code/AeonixWebsocketService/vendor/laravel/framework/src/Illuminate/Foundation/Console/Kernel.php(129): Illuminate\Console\Application->run()
#34 /home/vagrant/code/AeonixWebsocketService/artisan(37): Illuminate\Foundation\Console\Kernel->handle()
#35 {main}

* stopped the pause stream!
* Closing connection 0

请问各位大佬，能不能这样做？处理这种轮询请求的最佳方式是什么？

谢谢大家！

 1348  3 1

3个回答

2548a 2022-01-30

你这样死循环,系统根本不会把控制权还给workerman,所以woekerma后续也无法处理任何东西,所以肯定是不行的.但是我不了解你想实现的场景.你单纯说要不停轮询,这我没法给你建议

prcvoldermort 2022-01-30

谢谢您的回复。具体的场景是这样的，要轮询的服务器是一台电话软交换服务器，对外只提供了HTTP接口，需要轮询获取系统中最新的话机及话务状态，有一个getEvents请求到服务器后会有10秒钟的阻塞，如果10秒钟内服务器有新的话务事件或状态，此次请求会立即返回新的话务事件或状态，如果10秒钟内没有事件或状态，则返回一个通用的响应。无论有没有新的事件或状态，在上一个getEvents请求得到响应后，要立刻发起新的getEvents请求，如此循环下去，直到一个连接session的断开。
我也考虑过在onWorkerStart中使用定时器来定时发起getEvents请求，把新的连接session放入一个数组来遍历，但无法做到得到响应后再发送getEvents请求，导致产生了过多的请求。

2548a 2022-01-30

虽然你的需求我看的有点懵,但是你的处理方式是肯定不行的,按你这思路,一个人请求上来后,就一直卡在这轮询了,这个进程根本不可能再接收其它用户连接了.
我先说好,我不确定理解了你得逻辑,但是我肯定,不需要这样死循环肯定能处理得,只是你思维陷进误区了

// 我这只是写个简单思路,这里面肯定要携带你发起 新getEvents得参数,还有,里面也要做一个退出得条件
public function handle($client_id,$time){
    // 没有什么说是一秒不能停得,这种我只能说是你架构有问题
   Timer::add(1,function ()use($client_id,$time){
       $http = new Workerman\Http\Client();
       $http->get('https://example.com/', function($response) use($client_id,$time) {
           var_dump($response->getStatusCode());
           echo $response->getBody();
           $status = '';
           $now_time = time();

           // 有新状态
           if ($status){
               // 发送响应,
               Gateway::sendToClient( $client_id,  '');
               // 创建新链接

               $time = $now_time;
           }else{

               if ($now_time - $time > 10){
                   //10秒钟内没有事件或状态，返回一个通用的响应

                   //发起新请求
                   $time = $now_time;

               }else{
                   // 没达到10秒,继续轮询
               }
           }

           $this->handle($client_id,$time);

       }, function($exception)use($client_id,$time){
           echo $exception;

           $now_time = time();
           if ($now_time - $time > 10){
               //10秒钟内没有事件或状态，返回一个通用的响应

               //发起新请求
               $time = $now_time;

           }else{
               // 没达到10秒,继续轮询
           }

           $this->handle($client_id,$time);

       });
   },[],false);
}

public function onWebSocketConnect(string $client_id, array $data){
    $this->handle($client_id,time());
}