Ratchet PHP и длинные задачи

Question

Я использую ratchet php. Я начинаю это так:

$loop = \React\EventLoop\Factory::create(); 
$webSock = new \React\Socket\Server($loop); 

$webSock->listen($this->port, $this->host); 

$webServer = new \Ratchet\Server\IoServer(
    new \Ratchet\Http\HttpServer(
     new \Ratchet\WebSocket\WsServer(
      new PusherServer() 
     ) 
    ), 
    $webSock 
); 

return $loop; 

Теперь в моем onMessage() моего Pusherserver класса (который реализует MessageComponentInterface), я хочу, чтобы выполнить длинную, блокирующую задачу. Это будет HTTP-запрос, который может занять до десяти секунд.

Как я могу сделать onMessage() для обработки других запросов во время выполнения предыдущего HTTP-запроса? Я не могу использовать pthreads, поскольку у меня нет доступа к изменению версии php, которую я уже предоставил (что является потокобезопасным).

Answer 1

Это именно то, что вам нужно избегать при выполнении всего в вашем цикле событий: он не может блокироваться, потому что кто-то еще пытается подписаться или вызвать сообщение или что-нибудь еще управляемое событиями события не могут произойти, пока это не закончится.

Это больше проблема архитектуры, и как только вы выяснили, как это сделать, это упростит это и убедится, что оно работает для всех необходимых вам задач.

Ratchet обеспечивает привязку ZMQ - это потрясающе, потому что, как только вы его настроите, все, что вы получите на порте 5555, попадет в ваш цикл событий в onYourMethodName(), или как вы хотите это назвать!

Принимая это во внимание, вам необходимо отправить работу, которая должна быть выполнена в очередь на работу, другой процесс (ответ имеет расширение child-process, которое мне не особенно нравится, поскольку это опрос на земле пользователя, а не прерывания, управляемые вводом-выводом, как PHP PCNTL extension) или аналогичные.

Если вы хотите «просто заставить его работать», скройте работу, которая должна быть выполнена, вместе с идентификатором соединения или другим идентификатором, чтобы вы знали, к какому ответу нужно вернуться, в дочернем процессе и когда это будет отправлено. Это не будет блокировать!

Если вы хотите сделать это лучше, и я настоятельно рекомендую изучить это и архитектуру для этого, чтобы вы могли взять это знание с собой в своей карьере, когда вы снова подходите к асинхронной проблеме, подобной этой, подход «огонь-и-забыть». Огонь, что нужно сделать в очереди заданий в цикле событий, а затем забыть об этом.

Ваша очередь заданий может выполнять свою работу, и когда это будет сделано, запустите результат этого обратно через ZMQ (порт 5555, который прослушивает, помнит), который затем может отправить данные обратно клиенту.

Для удивительный разговор на очереди заданий, я настоятельно рекомендую this one от PHPNW.

Заключительное примечание, поскольку вы открываете эту вещь и слушаете порт 5555 для данных, вы можете отправить эти данные с в любом месте. Это может быть межпроцессное общение, так как в вашем приложении есть Java-приложение, которое отправляет данные на порт 5555 или буквально все. Это связывает вещи вместе, но не связывает их, что важно в вашей архитектуре.

---

Для примера фактического использования ZMQ, они обеспечивают все это на this page here (как связано выше), но я попытаюсь объяснить немного о том, что происходит.

$pull->bind('tcp://127.0.0.1:5555'); 
$pull->on('message', array($pusher, 'onYourMethodName')); 

Эта часть означает, что, когда ничего отправляет данные на порт 5555, и это «сообщение» (вы можете Google другие опции, доступные вместо сообщения), он будет называть onYourMethodName в вашем $pusher объекте. Это действительно настолько просто. Ничего более 5555, хитов $pusher::onYourMethodName.

В результате вам просто нужно создать свой метод в обработчике событий (рядом с onMessage(), onSubscribe() и т. Д.) ... Снова все это упоминается на этой странице.

public function onYourMethodName($data) 
{ 
    /** You'll probably want to send the data in JSON format **/ 
    /** Imagine you get through a 'topic' in here... **/ 
    $data = json_decode($data, true); 

    /** You should already have stored the people who are connected, topics etc - see the tutorial **/ 
    $topic = $this->subscribedTopics[$data['topic']]; 

    /** Send the data out to everyone subscribed to this topic **/ 
    $topic->broadcast($data); 
} 

Если вы хотите, чтобы отправлять данные конкретному пользователю, а не всем, существует много способов сделать это. Взгляните на this question на то, как я это сделал, но это было время назад.

Единственное, что вам нужно сделать сейчас, в вашем обработчике (в onMessage или что-то еще), на самом деле положить то, что нужно сделать в очереди, а также с кем он должен быть отправлен обратно (тема).

В конце своего рабочего делает это вещество и получать данные, нужно будет назвать это ударить код я показал выше:

$context = new ZMQContext(); 
$socket = $context->getSocket(ZMQ::SOCKET_PUSH, 'my pusher'); 
$socket->connect("tcp://localhost:5555"); 

$socket->send(json_encode($data)); 

Итак, вот что вам нужно сделать:

• Получить первый бит работает с прослушивает порт 5555
• Создать свой метод в обработчике событий и т.д.
• не беспокойтесь об очередях или что-нибудь еще только
• Создайте действительно простой php-скрипт, который выполняет вышеуказанные 4 строки кода, и докажите, что при запуске этого сценария отдельно он действительно отправляет данные в ваш цикл событий
• Затем подумайте о том, как добавить его в свой в очереди и попросите ваших работников справиться с этим