Swoole implementiert ein äußerst zuverlässiges Veröffentlichungs- und Abonnementsystem

$server = new SwooleServer('0.0.0.0', 8080, SWOOLE_PROCESS);
$server->set([
    'worker_num' => 2,
    'daemonize' => false,
]);
$server->on('WorkerStart', function($serv, $worker_id) {
    // 创建消息队列
    $queue_key = ftok(__FILE__, 'a');
    $queue = msg_get_queue($queue_key, 0666 | IPC_CREAT);
    // 将消息队列作为全局变量存放起来
    global $message_queue;
    $message_queue = $queue;
    // 启动消息处理进程
    if ($worker_id == 0) {
        $process = new SwooleProcess(function($process) {
            global $message_queue;
            while (true) {
                // 从消息队列中获取消息
                if (msg_receive($message_queue, 0, $msg_type, 1024, $msg, true, MSG_IPC_NOWAIT)) {
                    // 将消息发送给对应的订阅者
                    // TODO:实现发送消息的逻辑
                }
                // 隔一段时间循环一次
                usleep(100);
            }
        }, false, false);
        $process->start();
    }
});
$server->on('Connect', function($serv, $fd) {
    echo "Client[$fd]: Connect.
";
});
$server->on('Receive', function($serv, $fd, $from_id, $data) {
    global $message_queue;
    // 接收到消息，将消息存放到消息队列
    if (msg_send($message_queue, 1, $data, true, true)) {
        echo "Received message: $data
";
    } else {
        echo "Failed to send message to message queue.
";
    }
});
$server->on('Close', function($serv, $fd) {
    echo "Client[$fd]: Close.
";
});
$server->start();

$client = new SwooleClient(SWOOLE_SOCK_TCP);
if (!$client->connect('127.0.0.1', 8080)) {
    echo "Failed to connect to server.
";
    exit(1);
}
// 订阅主题
if (!$client->send("subscribe:topic1")) {
    echo "Failed to send subscribe message.
";
    exit(1);
}
// 接收消息
while (true) {
    $data = $client->recv();
    if ($data === false) {
        echo "Failed to receive message.
";
        break;
    }
    if (empty($data)) {
        continue;
    }
    echo "Received message: $data
";
}
$client->close();

$client = new SwooleClient(SWOOLE_SOCK_TCP);
if (!$client->connect('127.0.0.1', 8080)) {
    echo "Failed to connect to server.
";
    exit(1);
}
// 发布消息
if (!$client->send("publish:topic1:message1")) {
    echo "Failed to send publish message.
";
    exit(1);
}
$client->close();