tp8 使用rabbitMQ（4）路由模式

news2024/9/24 17:20:00

路由模式

在第三节中我们使用的交换机的 fanout 把生产者的消息广播到了所有与它绑定的队列中处理，但是我们能不能把特定的消息，发送给指定的队列，而不是广播给所有队列呢？
在这里插入图片描述
如图，交换机把 orange 类型的消息发送给了队列1处理，而带有 black 和 green标记的数据发送给了队列2来处理。
这时就要使用路由模式了

在路由模式中，要使用交换机的类型需要是直联模式，并且绑定的时候必须使用 route_key,而上节中使用的 fanout 模式会忽略这个值。

路由模式的使用方法很简单，就是在交换机和队列绑定的时候提供第三个参数 $routing_key

$channel->queue_bind($queue_name, $exchange_name, $routing_key);

代码和发布订阅模式的代码差不多，主要是 exchange的模式要改成直联，然后在消费者的代码中binding 时，指名 routing_key

生产者

<?php
declare (strict_types = 1);

namespace app\command;

use ba\Exception;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use think\console\Command;
use think\console\Input;
use think\console\input\Argument;
use think\console\input\Option;
use think\console\Output;

class RoutingMQProduce extends Command
{
    protected function configure()
    {
        // 指令配置
        $this->setName('routingmqproduce')
            ->setDescription('路由模式');
    }

    protected function execute(Input $input, Output $output)
    {
        //获取连接
        $connection = $this->getConnectRabbitMQ();
        //创建通道
        $channel = $connection->channel();
        //创建交换机
        /**
         * params exchange  自定义交换机名称
         * params type  交换机的类型，路由模式使用 直联 （direct）
         * params passive 是否消极声名
         * params durable 是否持久化
         * params auto_delete 是否自动删除
         * params internal 设置是否内置的, true表示是内置的交换器,客户端程序无法直接发送消息到这个交换器中, 只能通过交换器路由到交换器这个方式
         * params  nowait 相当于做一个异步版的声明,不等待返回，就让程序继续执行
         */
        $channel->exchange_declare("exchangeName","direct",false,false,false,false,false);
        //现在生产者只需要把消息发给交换机就可以了，所以不用在生产者中创建队列了（当然，想创建也是可以的）
        //在这里随机一个名称，来做为 routing_key

        for ($i = 0; $i < 20; $i++) {

            $routing_keys = ["orange","black","green"];
            shuffle($routing_keys);
            $routing_key = $routing_keys[0];


            $msgArr = [
                "name"=>"haha".$routing_key,  //这里把 routing＿key 传过去验证
                "age"=>'10',
                "sex"=>"female".$i
            ];
            $msg = new AMQPMessage(json_encode($msgArr),[
                "delivery_mode"=>AMQPMessage::DELIVERY_MODE_PERSISTENT
            ]);
            sleep(1);

            //这里发布时指定了 $routing_key
            $channel->basic_publish($msg,"exchangeName",$routing_key);
        }

        $channel->close();
        $connection->close();
    }

    protected function getConnectRabbitMQ(){
        try{
            $connection = new AMQPStreamConnection("192.168.3.228",'5672',"admin","123456");
            return $connection;
        }catch(Exception $e){
            throw new Exception("队列连接失败");
        }

    }
}

消费者代码

<?php
declare (strict_types = 1);

namespace app\command;

use ba\Exception;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use think\console\Command;
use think\console\Input;
use think\console\input\Argument;
use think\console\input\Option;
use think\console\Output;

class RoutingMQConsumer extends Command
{
    protected function configure()
    {
        // 指令配置
        $this->setName('routingmqconsumer')
            ->setDescription('路由模式的消费者');
    }

    protected function execute(Input $input, Output $output)
    {
        $connection = $this->connectRabbitMQ();
        $channel = $connection->channel();
        //创建两个队列
        $channel->queue_declare("queueName1",false,false,false,false,false);
        $channel->queue_declare("queueName2",false,false,false,false,false);
        //绑定交换机和队列，交换机的名称是在生产者中定义的
        $channel->queue_bind("queueName1","exchangeName","orange");
        $channel->queue_bind("queueName2","exchangeName","green");
        $channel->queue_bind("queueName2","exchangeName","black");
        //设置消息处理函数
        $callback1 = function($msg){
            $msgArr = json_decode($msg->body,true);
            echo "我是队列1，我只处理 orange 标记的数据  ".$msgArr["name"]."-11-".$msgArr["age"]."-11-".$msgArr["sex"].PHP_EOL;
            $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);   //这里让就是消息的应答了
        };
        $callback2 = function($msg){
            $msgArr = json_decode($msg->body,true);
            echo "我是队列2，我处理 black和green 标记的数据  ".$msgArr["name"]."-22-".$msgArr["age"]."-22-".$msgArr["sex"].PHP_EOL;
            $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);   //这里让就是消息的应答了
        };
        $channel->basic_consume("queueName1","",false,false,false,false,$callback1);
        $channel->basic_consume("queueName2","",false,false,false,false,$callback2);
        while(count($channel->callbacks)){
            $channel->wait();
        }
    }

    protected function connectRabbitMQ(){
        try{
            $connection = new AMQPStreamConnection("192.168.3.228",'5672',"admin","123456");
            return $connection;
        }catch(Exception $e){
            throw new Exception("队列连接失败");
        }

    }
}