>[info] fanout 订阅/广播模式 交换器会把所有发送到该交换器的消息路由到所有与该交换器绑定的消息队列中。**订阅模式与Binding Key和Routing Key无关**，交换器将接受到的消息分发给有绑定关系的所有消息队列队列（不论Binding Key和Routing Key是什么）。类似于子网**广播**，子网内的每台主机都获得了一份复制的消息。**Fanout交换机转发消息是最快的**。   1. `rabbitmq.php` 配置文件信息 ~~~ <?php // rabbitmq 配置信息 return [ # 连接信息 'amqp' => [ 'host' => '127.0.0.1', 'port'=>'5672', 'user'=>'guest', 'password'=>'guest', 'vhost'=>'/' ], # 扇出队列 'fanout_queue' => [ 'exchange_name' => 'fanout_exchange', 'exchange_type'=>'fanout',# 订阅/广播模式 'queue_name' => 'fanout_queue', 'route_key' => 'fanout_route_key', 'consumer_tag' => 'fanout' ], # 商品队列 'goods_queue' => [ 'exchange_name' => 'goods_fanout_exchange', 'exchange_type'=>'fanout',# 订阅/广播模式 'queue_name' => 'goods_queue', 'route_key' => 'goods_route_key', 'consumer_tag' => 'goods' ], # 订单队列 'order_queue' => [ 'exchange_name' => 'order_fanout_exchange', 'exchange_type'=>'fanout',# 订阅/广播模式 'queue_name' => 'order_queue', 'route_key' => 'order_route_key', 'consumer_tag' => 'order' ], ]; ~~~ 2. `Consumer` 消费者 ~~~ <?php declare (strict_types = 1); namespace app\common\service\rabbitmq; //use app\common\model\test\MessageModel; use PhpAmqpLib\Connection\AMQPStreamConnection; use think\facade\Log; /** * 消费者 */ class Consumer { /** * 商品消费者 * @return \think\Response */ public function goods() { $mqConfig = config('rabbitmq.amqp'); $goodsConfig = config('rabbitmq.goods_queue'); // 创建连接 $connection = new AMQPStreamConnection( $mqConfig['host'], $mqConfig['port'], $mqConfig['user'], $mqConfig['password'], $mqConfig['vhost'] ); // 连接信道 $channel = $connection->channel(); // 设置消费者（Consumer）客户端同时只处理一条队列 // 这样是告诉RabbitMQ，再同一时刻，不要发送超过1条消息给一个消费者（Consumer）， // 直到它已经处理了上一条消息并且作出了响应。这样，RabbitMQ就会把消息分发给下一个空闲的消费者（Consumer）。 // 消费者端要把自动确认autoAck设置为false，basic_qos才有效。 // 流量控制 $channel->basic_qos(0, 1, false); // 同样是创建路由和队列，以及绑定路由队列，注意要跟producer（生产者）的一致 // 这里其实可以不用设置，但是为了防止队列没有被创建所以做的容错处理 // 创建交换机(Exchange) $channel->exchange_declare($goodsConfig['exchange_name'], $goodsConfig['exchange_type'], false, true, false); // 创建队列 $channel->queue_declare($goodsConfig['queue_name'], false, true, false, false); // 绑定队列和交换机 $channel->queue_bind($goodsConfig['queue_name'], $goodsConfig['exchange_name']); /** * 消费消息 * * queue: queue_name 被消费的队列名称 * consumer_tag: consumer_tag 消费者客户端身份标识，用于区分多个客户端 * no_local: false 这个功能属于AMQP的标准，但是RabbitMQ并没有做实现 * no_ack: true 收到消息后，是否不需要回复确认即被认为被消费 * exclusive: false 是否排他，即这个队列只能由一个消费者消费。适用于任务不允许进行并发处理的情况下 * nowait: false 不返回执行结果，但是如果排他开启的话，则必须需要等待结果的，如果两个一起开就会报错 * callback: $callback 回调逻辑处理函数 */ $channel->basic_consume($goodsConfig['queue_name'], $goodsConfig['consumer_tag'], false, false, false, false, array($this, 'process_message')); // 退出，执行shutdown来关闭通道与连接 register_shutdown_function(array($this, 'shutdown'), $channel, $connection); // 阻塞队列监听事件 while (count($channel->callbacks)) { $channel->wait(); } } /** * 订单消费者 * @return \think\Response */ public function order() { $mqConfig = config('rabbitmq.amqp'); $orderConfig = config('rabbitmq.order_queue'); // 创建连接 $connection = new AMQPStreamConnection( $mqConfig['host'], $mqConfig['port'], $mqConfig['user'], $mqConfig['password'], $mqConfig['vhost'] ); // 连接信道 $channel = $connection->channel(); // 设置消费者（Consumer）客户端同时只处理一条队列 // 这样是告诉RabbitMQ，再同一时刻，不要发送超过1条消息给一个消费者（Consumer）， // 直到它已经处理了上一条消息并且作出了响应。这样，RabbitMQ就会把消息分发给下一个空闲的消费者（Consumer）。 // 消费者端要把自动确认autoAck设置为false，basic_qos才有效。 // 流量控制 $channel->basic_qos(0, 1, false); // 同样是创建路由和队列，以及绑定路由队列，注意要跟producer（生产者）的一致 // 这里其实可以不用设置，但是为了防止队列没有被创建所以做的容错处理 // 创建交换机(Exchange) $channel->exchange_declare($orderConfig['exchange_name'], $orderConfig['exchange_type'], false, true, false); // 创建队列 $channel->queue_declare($orderConfig['queue_name'], false, true, false, false); // 绑定队列和交换机 $channel->queue_bind($orderConfig['queue_name'], $orderConfig['exchange_name']); /** * 消费消息 * * queue: queue_name 被消费的队列名称 * consumer_tag: consumer_tag 消费者客户端身份标识，用于区分多个客户端 * no_local: false 这个功能属于AMQP的标准，但是RabbitMQ并没有做实现 * no_ack: true 收到消息后，是否不需要回复确认即被认为被消费 * exclusive: false 是否排他，即这个队列只能由一个消费者消费。适用于任务不允许进行并发处理的情况下 * nowait: false 不返回执行结果，但是如果排他开启的话，则必须需要等待结果的，如果两个一起开就会报错 * callback: $callback 回调逻辑处理函数 */ $channel->basic_consume($orderConfig['queue_name'], $orderConfig['consumer_tag'], false, false, false, false, array($this, 'process_message')); // 退出，执行shutdown来关闭通道与连接 register_shutdown_function(array($this, 'shutdown'), $channel, $connection); // 阻塞队列监听事件 while (count($channel->callbacks)) { $channel->wait(); } } /** * 消息处理 * @param $message */ public function process_message($message) { if ($message->body !== 'quit') { $messageBody = json_decode($message->body); // 自定义的消息类型 if (!isset($messageBody->message_type)) { Log::write("error data:" . $message->body, 2); } else { // $messageModel = new MessageModel(); try { // 消息 Log::write("message_data:" . json_encode($message, JSON_UNESCAPED_UNICODE)); $body = json_decode($message->body, true); dump("消费消息如下："); dump($body); // $messageModel->test($body); } catch (\Think\Exception $e) { Log::write($e->getMessage(), 2); Log::write(json_encode($message), 2); } catch (\PDOException $pe) { Log::write($pe->getMessage(), 2); Log::write(json_encode($message), 2); } } } // 手动确认ack，确保消息已经处理 $message->delivery_info['channel']->basic_ack($message->delivery_info['delivery_tag']); // Send a message with the string "quit" to cancel the consumer. if ($message->body === 'quit') { $message->delivery_info['channel']->basic_cancel($message->delivery_info['consumer_tag']); } } /** * 关闭进程 * @param $channel * @param $connection */ public function shutdown($channel, $connection) { // 关闭信道 $channel->close(); // 关闭连接 $connection->close(); } } ~~~ 3. `Producer` 生产者 ~~~ <?php declare (strict_types = 1); namespace app\common\service\rabbitmq; use PhpAmqpLib\Connection\AMQPStreamConnection; use PhpAmqpLib\Message\AMQPMessage; /** * rabbitmq 生产者 */ class Producer { // 连接 protected $connection; // 管道 protected $channel; // 配置内容（基本） protected $mqConfig; // 配置内容（扇出） protected $fanoutConfig; public function __construct() { $this->mqConfig = config('rabbitmq.amqp'); $this->fanoutConfig = config('rabbitmq.fanout_queue'); // 创建连接 $this->connection = new AMQPStreamConnection( $this->mqConfig['host'], $this->mqConfig['port'], $this->mqConfig['user'], $this->mqConfig['password'], $this->mqConfig['vhost'] // 虚拟主机（起到消息隔离的作用） ); // 创建通道 $this->channel = $this->connection->channel(); } /** * 发送消息 * @param $data 消息内容 */ public function send($data) { /* * 流量控制 Specifies QoS * 消费者在开启acknowledge的情况下，对接收到的消息需要异步对消息进行确认 * 由于消费者自身处理能力有限，从rabbitmq获取一定数量的消息后，希望rabbitmq不再将队列中的消息推送过来， * 当对消息处理完后（即对消息进行了ack，并且有能力处理更多的消息）再接收来自队列的消息 * @param int $prefetch_size 最大unacked消息的字节数 * @param int $prefetch_count 最大unacked消息的条数 * @param bool $a_global 上述限制的限定对象，false限制单个消费者，true限制整个通道 * @return mixed */ $this->channel->basic_qos(0, 1, false); /** * 创建队列(Queue) * name: hello 队列名称 * passive: false 如果设置true存在则返回OK，否则就报错。设置false存在返回OK，不存在则自动创建 * durable: true 是否持久化，设置false是存放到内存中的，RabbitMQ重启后会丢失;设置true,则代表是一个持久化的队列，服务重启后也会存在，因为服务会把持久化的queue存放到磁盘上当服务重启的时候，会重新加载之前被持久化的queue * exclusive: false 是否排他，指定该选项为true则队列只对当前连接有效，连接断开后自动删除 * auto_delete: false 是否自动删除，当最后一个消费者断开连接之后队列是否自动被删除 */ $this->channel->queue_declare($this->fanoutConfig['queue_name'], false, true, false, false); /** * 创建交换机(Exchange) * name: hello 交换机名称 * type: direct 交换机类型，分别为direct/fanout/topic，参考另外文章的Exchange Type说明。 * passive: false 如果设置true存在则返回OK，否则就报错。设置false存在返回OK，不存在则自动创建 * durable: false 是否持久化，设置false是存放到内存中的，RabbitMQ重启后会丢失 * auto_delete: false 是否自动删除，当最后一个消费者断开连接之后队列是否自动被删除 */ $this->channel->exchange_declare($this->fanoutConfig['exchange_name'], $this->fanoutConfig['exchange_type'], false, true, false); // 绑定消息交换机和队列 $this->channel->queue_bind($this->fanoutConfig['queue_name'], $this->fanoutConfig['exchange_name'],$this->fanoutConfig['route_key']); // 将要发送数据变为json字符串 $messageBody = json_encode($data, JSON_UNESCAPED_UNICODE); /** * 创建AMQP消息类型 * delivery_mode 消息是否持久化 * AMQPMessage::DELIVERY_MODE_NON_PERSISTENT 不持久化 * AMQPMessage::DELIVERY_MODE_PERSISTENT 持久化 */ $message = new AMQPMessage($messageBody, [ 'content_type' => 'text/plain', 'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT ]); /** * 发送消息 * msg: $message AMQP消息内容 * exchange: vckai_exchange 交换机名称 * routing_key: hello 路由key */ $this->channel->basic_publish($message, $this->fanoutConfig['exchange_name'], $this->fanoutConfig['route_key']); // 关闭连接 $this->stop(); } // 关闭连接 public function stop() { $this->channel->close(); $this->connection->close(); } } ~~~ 4. 创建 `goods` 和 `order` 自定义命令，作为消费者。 ~~~php php think make:command Goods ~~~ ~~~ <?php declare (strict_types = 1); namespace app\command; use think\console\Command; use think\console\Input; use think\console\input\Argument; use think\console\input\Option; use think\console\Output; use app\common\service\rabbitmq\Consumer; class Goods extends Command { protected function configure() { // 指令配置 $this->setName('goods') ->setDescription('the goods command'); } protected function execute(Input $input, Output $output) { // 启动消费者 $consumer = new Consumer(); $consumer->goods(); } } ~~~ ``` php think make:command Order ``` ~~~ <?php declare (strict_types = 1); namespace app\command; use think\console\Command; use think\console\Input; use think\console\input\Argument; use think\console\input\Option; use think\console\Output; use app\common\service\rabbitmq\Consumer; class Order extends Command { protected function configure() { // 指令配置 $this->setName('order') ->setDescription('the order command'); } protected function execute(Input $input, Output $output) { // 启动消费者 $consumer = new Consumer(); $consumer->order(); } } ~~~ 5. 配置 `config/console.php` 命令行 ~~~ <?php // +---------------------------------------------------------------------- // | 控制台配置 // +---------------------------------------------------------------------- return [ // 指令定义 'commands' => [ // rabbitMq 调用消费者 'consumer' => 'app\command\Consumer', // 商品 'goods' => 'app\command\Goods', // 订单 'order' => 'app\command\Order', ], ]; ~~~ 6. 路由发送消息 `route/app.php` ~~~ Route::get('rabbitmq', function (){ $producer = new \app\common\service\rabbitmq\Producer(); $data = [ 'message_type' => 2, 'order_id' => 3, 'user_id' => 3, 'time' => time(), 'message' => "发送的消息内容：您的快递已到的配送站。（" . rand(1,100) . ")" ]; $producer->send($data); }); ~~~ 7. 启动消费者 ``` php think order ``` ``` php think goods ``` 8. 生产者发送消息