协程安全的进程通讯 With Swoole

swoole-process-socket-with-Coroutine.php

<?php
Swoole\Runtime::enableCoroutine();
$processContainer = [];
for ($i = 0; $i < 2; $i++) {
    $process                = new \Swoole\Process(function (\Swoole\Process $workerProcess) {
        \Swoole\Coroutine::create(function () use ($workerProcess) {
            while (true) {
                $socket = $workerProcess->exportSocket();
                $pid    = posix_getpid();
                echo 'Idle:' . $pid . PHP_EOL;
                $socket->send(serialize(['type' => 'idle']));//Tell master that I'm Idle
                $receive     = ($socket->recv());
                $receivedLen = strlen($receive);
                $_           = unserialize($receive);
                printf('%s @%s: Received %s,Type %s' . PHP_EOL, date('Y-m-d H:i:s'), posix_getpid(), $receivedLen, gettype($_));
                \co::sleep(5);
                $socket->send(serialize(['type' => 'return', 'payload' => $receive]));//Tell master that I'm Idle
            };
        });
        \Swoole\Coroutine::create(function () {
            while (true){
                echo posix_getpid() . ': Coroutine Alive Not Block By Sleep :)' . PHP_EOL;
                \co::sleep(1);
            }
        });
        \Swoole\Timer::tick(1*1000,function (){
                echo posix_getpid() . ': Coroutine Alive Not Block  By Time Ticker :))' . PHP_EOL;
        });
    }, false, SOCK_DGRAM, true);
    $pid                    = $process->start();
    $processContainer[$pid] = $process;
}
foreach ($processContainer as $pid => $process) {
    Swoole\Coroutine::create(function () use ($process) {
        $socket = $process->exportSocket();
        while (true) {
            $retOrgi = $socket->recv();
            $ret     = unserialize($retOrgi);
            if ('idle' === $ret['type'] ?? '') {
                $t = str_repeat('t', 65535 - 64 - 1);
                echo '-------' . strlen($t) . '-------' . PHP_EOL;
                $send = $socket->send(serialize($t));//MAX is 65481 b ?
                echo 'SENDED: ' . $send . PHP_EOL;
            } elseif ('return' === $ret['type'] ?? '') {
                //               echo 'Got Return:'.var_export($ret['payload'],true).PHP_EOL;
                echo 'Got Return' . PHP_EOL;
            } else {
                echo 'illegal data:' ./*$retOrgi.*/
                    PHP_EOL;
            }
        };
    });
}

echo 'Done' . PHP_EOL;
//$processContainer[0]->exportSocket()->send('xxxxxxxx_'.'999');

\Swoole\Process::signal(SIGCHLD, function ($sig) {
    //todo remove this when quiting
    //必须为false，非阻塞模式
    while ($ret = \Swoole\Process::wait(false)) {
        var_export($ret);
        echo "PID={$ret['pid']}\n";
        $this->rebootProcess($ret);
    }
});

swoole/swoole-src#2527 (comment)