6

这个问题与使用 react\stomp 仅消耗来自 RabbitMQ 的 N 条消息非常相关,分别确认它们然后退出,但更笼统一些。

例如,我有一个缓慢的 I/O 操作:

$port = 4000;

$loop = React\EventLoop\Factory::create();
$socket = new React\Socket\Server($loop);
$socket->on('connection', function ($conn) use ($loop){

    $conn->on('data', function ($data) use ($conn, $loop) {

        if ($data == 42) {
            // this instantly stop the loop
            $loop->stop();
        }

        $process = new React\ChildProcess\Process('sleep 5; echo ' . $data);

        $loop->addTimer(0.001, function($timer) use ($process, $conn) {
            $process->start($timer->getLoop());

            $process->stdout->on('data', function($output) use ($conn) {
                if ($output) {
                    $conn->write("> $output");
                }
            });
        });
    });
});
echo "Socket server listening on port $port.\n";
echo "You can connect to it by running: telnet localhost $port\n";
$socket->listen($port);
$loop->run();
echo "exited";

当我运行时,$loop->run();我想在某个时候停止它,例如通过计时器,在接受 N 个请求或任何其他事件(如pcntl_signal,或数据断言)之后,如示例中所示。

挑战是在退出之前完成所有开始的工作,我不知道如何实现它。

在服务器控制台中,我有:

Socket server listening on port 4000.
You can connect to it by running: telnet localhost 4000
exited

在客户端控制台中,我有:

Trying 127.0.0.1...
Connected to localhost.
Escape character is '^]'.
1
2
3
4
5
> 1
> 2
42
Connection closed by foreign host.

其中 1,2,3,4,5 以 1 秒的间隔输入

相反,我希望看到这样的事情:

Trying 127.0.0.1...
Connected to localhost.
Escape character is '^]'.
1
2
3
4
5
> 1
> 2
42
> 3
6
> 4
> 5
Connection closed by foreign host.
4

1 回答 1

1

正如评论中提到的,要优雅地退出,您需要跟踪正在运行的进程并仅在所有进程完成时才停止循环:

$socket->on('connection', function ($conn) use ($loop) {
    $processes = new SplObjectStorage();
    $stop = false;

    $conn->on('data', function ($data) use ($conn, $loop, $processes, &$stop) {
        if ('42' === trim($data)) {
            $stop = true;
            if (!$processes->count()) {
                $loop->stop();
            }
        }

        if ($stop) {
            return;
        }

        $process = new React\ChildProcess\Process('sleep 5; echo ' . $data);
        $processes->attach($process);

        $process->on('exit', function () use ($process, $processes, &$stop, $loop) {
            $processes->detach($process);

            if ($stop && !$processes->count()) {
                $loop->stop();
            }
        });

        $loop->addTimer(0.001, function($timer) use ($process, $conn) {
            $process->start($timer->getLoop());

            $process->stdout->on('data', function($output) use ($conn) {
                if ($output && '42' !== trim($output)) {
                    $conn->write("> $output");
                }
            });
        });
    });
});
于 2015-10-06T10:22:26.943 回答