多线程编程的利器：alexanderc/threadator库的实践与应用

composer require alexanderc/threadator:dev-master

<?php
require '/path/to/vendor/autoload.php';

$runtime = new \Threadator\Runtime();
$factory = new \Threadator\Factory($runtime);

$communication = \Threadator\Communication\Communication::create($runtime, 'msgQueue');
$runtime->setCommunication($communication);

for($i = 0; $i < 5; $i++) {
    $thread = $factory->createCallable(function($thread) {
        $mutex = $thread->createMutex("echo", \Threadator\Mutex::T_FUNCTION);
        $mutex->waitAcquire();
        sleep(mt_rand(1, 3));
        echo "Running Thread #{$thread->getPid()}...\n";
        $thread->receiveMessage($message);
        $thread->sendMessage("#{$thread->getPid()}: {$message}");
    });
}

echo "Main process #{$runtime->getPid()} running!\n";

$runtime->run();

foreach($runtime->broadcastMessage(microtime(true)) as list($result, $thread)) {
    echo "Result for msg #{$thread->getPid()} -> {$result}\n";
}

$messages = [];
foreach($runtime->receiveMessage() as $result => $message) {
    if($result) {
        $messages[] = $message;
    }
}
echo "Thread messages: " . implode(", ", $messages) . "\n";

$runtime->join();

exit("Main process #{$runtime->getPid()} stopped!\n");

<?php
use Threadator\Communication\Driver\ADriver;

class TestDriver extends ADriver
{
    protected function init()
    {
        // 初始化方法
    }

    public function send($key, $message)
    {
        // 发送消息方法
    }

    public function touch($key, & $message)
    {
        // 尝试获取消息，但不阻塞
    }

    public function receive($key, & $message)
    {
        // 阻塞直到第一个消息到达
    }
}