【RabbitMq】Direct（Routing路由选择）模型、Topic（动态路由）模型

野性酷女 2022-10-28 10:12 174阅读 0赞

# Direct（路由选择）订阅消息模型 #

在`fanout`消息模型中，一条消息会被所有已订阅的消费者消费。而在某些情况下，我们希望这个消息被特定的消费者消费。此时就可以用`exchange`中的`Direct`类型来实现，在`fanout`中，消息的转发会直接忽略`Routing Key`，而在`Derict`中，则会将消息自带的`Routing Key`与`exchange`的`Routing Key`相比较，相等则发送到该交换机绑定的队列。

**在direct模型下**

*  消息发送方发送到`exchange`时需要给消息携带一个`Routing key`
 *  `exchange`交换机不能将消息分发给所有的消费者了，`exchange`需要比较消息中的`Routing key`与自己路由中`Ruoting key`是否相等，相等的话则将消息发送给绑定的队列进行消费
 *  队列与交换机绑定不再是任意绑定，而需要指定`Routing key`

![Direct][]

*  P：`Provider`（消息生产者），将消息发送给`exchange`时给消息设置一个`Routing key`
 *  X：`exchange`（交换机），收到消息后将消息的`routing key`与自己的虚拟表去比对，相等则将消息转发给对应绑定的队列
 *  C1：只接受`Routing Key`为`error`的消息
 *  C2：只接受`Routing Key`为`info`或着`error`或着`warning`的消息

## 生产者 ##

该生产者发送`routingkey` 为`warning`的消息

public class Provider { 
        public static void main(String[] args) throws IOException { 
            Connection connection = RabbitMqUtil.getConnection();
            Channel channel = connection.createChannel();
            //通道声明交换机
            channel.exchangeDeclare("logs_direct","direct");
            String routingkey = "warning";
            //生产消息
            channel.basicPublish("logs_direct",routingkey,null,"基于direct模型的路由选择消息发送".getBytes());
            RabbitMqUtil.close(channel,connection);
        }
    }

## 多消费者 ##

消费者1只消费`routingkey`为`error`的消息

**customer1：**

public class Customer1 { 
        public static void main(String[] args) throws IOException { 
            Connection connection = RabbitMqUtil.getConnection();
            Channel channel = connection.createChannel();
            //通道绑定交换机
            channel.exchangeDeclare("logs_direct","direct");
            //创建一个临时队列
            String queue = channel.queueDeclare().getQueue();
            //声明route key
            String routingkey = "error";
            //队列绑定交换机,指明只接受消费route key为error的消息
            channel.queueBind(queue,"logs_direct",routingkey);
            //消费
            channel.basicConsume(queue,true,new DefaultConsumer(channel){ 
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                    System.out.println("消费者一消费route key为："+routingkey+"|"+"消息为："+new String(body));
                }
            });
        }
    }

消费者2只消费`routingkey`为`error`、`warning`、`info`的消息

**customer2：**

public class Customer2 { 
        public static void main(String[] args) throws IOException { 
            Connection connection = RabbitMqUtil.getConnection();
            Channel channel = connection.createChannel();
            //通道绑定交换机
            channel.exchangeDeclare("logs_direct","direct");
            //创建一个临时队列
            String queue = channel.queueDeclare().getQueue();
            //队列绑定交换机,指明只接受消费route key为error的消息
            channel.queueBind(queue,"logs_direct","error");
            channel.queueBind(queue,"logs_direct","warning");
            channel.queueBind(queue,"logs_direct","info");
            //消费
            channel.basicConsume(queue,true,new DefaultConsumer(channel){ 
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                    System.out.println("消费者二消费："+new String(body));
                }
            });
        }
    }

**消费结果：**

`customer1`：  
![customer1][]  
`customer2`：  
![customer2][]

# Topic（动态路由模型） #

![Topic][]  
该模型在Direct模型上增加了两个通配符，在消费者端指定接收的`routekey`时可直接模糊匹配即可。

> ‘\*’ (star) can substitute for exactly one word.  
> ‘\#’ (hash) can substitute for zero or more words.

channel.queueBind(queue,"topics","kexing.site.#");

指定`kexing.site`后面可以跟一个或着多个单词，比如`kexing.site.happy.day`，也可以为空

[Direct]: /images/20221024/2a412afe4dca4777ad4db89be3b1e1fe.png
[customer1]: /images/20221024/09648595defc4d60ad603eb8f195c41d.png
[customer2]: /images/20221024/21a32998e45346918d9116856cb055d5.png
[Topic]: /images/20221024/4b78776bd9484f698ad33c3605e3a1b7.png