概述

workerman/rabbitmq 是一个异步RabbitMQ客户端，使用AMQP协议。

RabbitMQ是一个基于AMQP（高级消息队列协议）实现的开源消息组件，它主要用于在分布式系统中存储和转发消息。RabbitMQ由高性能、高可用以及高扩展性出名的Erlang语言写成，具有高度的可靠性和可扩展性。它支持多种消息协议，包括AMQP、STOMP、MQTT等，并广泛应用于消息队列、消息中间件等领域。

RabbitMQ允许应用程序通过消息传递进行通信，这使得不同的应用程序可以在不同的语言和操作系统之间进行通信。

RabbitMQ的消息工作机制涉及消息从发送端到接收端的流转过程。在这个过程中，消息首先被发送到交换机（Exchange），然后交换机根据路由规则将消息路由到一个或多个队列（Queue）中。消费者（Consumer）从队列中获取消息并进行处理。

生产者和消费者

安装

composer require workerman/rabbitmq

消费者

receive.php

<?phpdeclare(strict_types=1);use Bunny\Channel;
use Bunny\Message;
use Workerman\Worker;
use Workerman\RabbitMQ\Client;require_once __DIR__ . '/vendor/autoload.php';$worker = new Worker();
$worker->eventLoop = \Workerman\Events\Revolt::class;$worker->onWorkerStart = function() {// Create RabbitMQ Client$client = Client::factory(['host' => '127.0.0.1','port' => 5672,'user' => 'guest','password' => 'guest','vhost' => '/','heartbeat' => 60,'heartbeat_callback' => function () {echo " [-] coroutine-consumer-heartbeat\n";},'interval' => [100, 300]])->connect();$channel = $client->channel();$channel->queueDeclare('hello-coroutine');// Consumer$channel->consume(function (Message $message, Channel $channel, \Bunny\AbstractClient $client) {echo " [>] Received ", $message->content, "\n";},'hello-coroutine','',false,true);$client->run();echo ' [*] Waiting for messages. To exit press CTRL+C', "\n";// Producer\Workerman\Timer::add($interval = 5 , function () use ($channel) {$channel->publish($message = 'Hello World By Self Timer. ' . time(), [], '', 'hello-coroutine');echo " [<] Sent $message\n";});echo " [!] Producer timer created, interval: $interval s.\n";};
Worker::runAll();

运行命令

php receive.php start

基于 Workerman 发布

send.php

<?phpdeclare(strict_types=1);use Workerman\RabbitMQ\Client;
use Workerman\Worker;require_once __DIR__ . '/vendor/autoload.php';$worker = new Worker();
$worker->eventLoop = \Workerman\Events\Revolt::class;$worker->onWorkerStart = function() {$client = Client::factory(['host' => 'host.docker.internal','port' => 5672,'user' => 'guest','password' => 'guest','vhost' => '/','heartbeat' => 60,'heartbeat_callback' => function () {echo "coroutine-producer-heartbeat\n";}])->connect();$channel = $client->channel();$channel->queueDeclare('hello-coroutine');// 每5秒发一个消息\Workerman\Timer::add(5, function () use ($channel) {$channel->publish($message = 'Hello World By Workerman Env Producer. ' . time(), [], '', 'hello-coroutine');echo " [x] Sent '$message'\n";});
};
Worker::runAll();

运行命令

php send.php start

基于 PHP-FPM 发布

script.php

<?phpdeclare(strict_types=1);use Workerman\RabbitMQ\Client;require_once __DIR__ . '/vendor/autoload.php';$client = Client::factory(['host' => 'host.docker.internal','port' => 5672,'user' => 'guest','password' => 'guest','vhost' => '/','heartbeat' => 60,'heartbeat_callback' => function () {echo "coroutine-producer-heartbeat\n";}
])->connect();
$channel = $client->channel();
$channel->queueDeclare('hello-coroutine');
$res = $channel->publish($message = 'Hello World By Normal Producer. ' . time(), [], '', 'hello-coroutine');echo " [x] Sent '$message', success: $res\n";

运行命令

php script.php

异步消费者

receive.php

<?phpuse Bunny\Channel;
use Bunny\Message;
use Workerman\Worker;
use Workerman\RabbitMQ\Client;require __DIR__ . '/vendor/autoload.php';$worker = new Worker();$worker->onWorkerStart = function() {(new Client())->connect()->then(function (Client $client) {return $client->channel();})->then(function (Channel $channel) {return $channel->queueDeclare('hello', false, false, false, false)->then(function () use ($channel) {return $channel;});})->then(function (Channel $channel) {echo ' [*] Waiting for messages. To exit press CTRL+C', "\n";$channel->consume(function (Message $message, Channel $channel, Client $client) {echo " [x] Received ", $message->content, "\n";},'hello','',false,true);});
};
Worker::runAll();

运行命令

php receive.php start

异步生产者

send.php

<?php
use Bunny\Channel;
use Bunny\Message;
use Workerman\Worker;
use Workerman\RabbitMQ\Client;require __DIR__ . '/vendor/autoload.php';$worker = new Worker();$worker->onWorkerStart = function() {(new Client())->connect()->then(function (Client $client) {return $client->channel();})->then(function (Channel $channel) {return $channel->queueDeclare('hello', false, false, false, false)->then(function () use ($channel) {return $channel;});})->then(function (Channel $channel) {echo " [x] Sending 'Hello World!'\n";return $channel->publish('Hello World!', [], '', 'hello')->then(function () use ($channel) {return $channel;});})->then(function (Channel $channel) {echo " [x] Sent 'Hello World!'\n";$client = $channel->getClient();return $channel->close()->then(function () use ($client) {return $client;});})->then(function (Client $client) {$client->disconnect();});
};
Worker::runAll();

运行命令

php send.php start