RasbbitMQ 交换机、路由键与队列绑定

比眉伴天荒 2022-11-13 10:25 164阅读 0赞

*  交换器Exchange

我们之前讲述的是将消息投递到队列中再有消费者区消费，实际上RabbitM的运作流程不是这样的。真实情况是，生产者将消息发送到 Exchange (交换器：“X”)，由交换器将消息路由到一个或者多个队列中，消费者再去消费消息。如果路由不到，或许会返回给生产者，或许直接丢弃（这个看具体配置后面会讲到：）。  
首先明白两个概念：  
BindingKey：绑定键，交换机都是需要与队列进行绑定，这里交换键可以简单的理解为交换器与队列之间的路径的名称（可以重复，即可以把多条队列以同一绑定键与路由器绑定）  
RoutingKey：路由键，生产者发送消息的时候可以带上路由键发送给交换器，交换器就会根据路由键去匹配队列（路由键 与交换器的匹配）

*  再了解常用的三种交换器

direct：当消息中的RoutingKey（路由键）和交换器与队列之间的BindingKey（绑定键）完全匹配时，会发送消息给相应的队列  
fanout：把当前路由器下所有消息发送到与该交换器绑定的队列中（不在意绑定键的有无与匹配）  
topic ：将消息路由到路由键与绑定键相匹配的队列中。

*  匹配规则：

路由键一般：单词.单词.单词 构成.（单词为一个或者几个字母组成）  
采用"\*“和”\#"，用于做模糊匹配,其中"\#“用于匹配一个单词，”"用于匹配多个单词(0~n)个。  
举例说明：路由键：.kkk.\*可以匹配绑定键：rrr.kkk.lll。路由键：\#.jjj.\# 可以匹配绑定键：xxx.mmm.kkk

*  RabbitMQ路由器运作流程

先声明一个路由器  
//第一个参数为路由器名字，第二个为路由器类型，第三个是否持久化，第四个设置是否自动删除，第五个携带参数  
channel.exchangeDeclare(EXCHANGE\_NAME,“direct”,true,false,null);  
生产者将路由器与队列绑定，再为绑定的路径赋值一个绑定键  
//第一参数为队列名，第二个参数为路由器，第三个参数为绑定键  
channel.queueBind(QUEUE\_NAME,EXCHANGE\_NAME,BINDING\_KEY);  
发送消息时指定路由，以及绑定路由键，携带参数，消息体  
channel.basicPublish(EXCHANGE\_NAME, ROUTING\_KEY, null, message.getBytes());

*  生产者代码

以下代码声明了一个路由器，并且声明了两个队列，将两个队列与路由器进行绑定，并且赋值了同一绑定键，最后向路由器中发送消息，携带路由键。这个时候路由器会将消息的路由键与路由器和队列绑定说产生的所有绑定键进行对比，会向匹配上的队列区投放消息，每个队列都会投放

public class Send {
        //队列名
        private final static String QUEUE_NAME = "wsd_test";
        //队列2名
        private final static String QUEUE_NAME_SECOND = "wsd_test_second";
        //路由器名
        private final static String EXCHANGE_NAME = "exchange";
        //绑定键
        private final static String BINDING_KEY = "wsd_exchange";
        //路由键
        private final static String ROUTING_KEY = "wsd_exchange";
        
        private static Connection connection =null;
        private static Channel channel = null;
        public static void main(String[] args) {
            try{
                // 获取到连接以及mq通道
                connection = ConnectionUtil.getConnection();
                // 从连接中创建通道
               channel = connection.createChannel();
               
               //声明了一个direct 类型的交换器
               channel.exchangeDeclare(EXCHANGE_NAME,"direct",true,false,null);
               
                 // 声明队列QUEUE_NAME
                channel.queueDeclare(QUEUE_NAME, false, false, false, null);
                // 声明队列QUEUE_NAME_SECOND
                channel.queueDeclare(QUEUE_NAME_SECOND, false, false, false, null);
                
                //将路由与队列绑定，再为绑定的路径赋值一个绑定键
               channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,BINDING_KEY);
               //将路由与队列绑定，再为绑定的路径赋值一个绑定键
               channel.queueBind(QUEUE_NAME_SECOND,EXCHANGE_NAME,BINDING_KEY);
    
                //发送数据
                for (int i=0;i<10;i++){
                    // 消息内容
                    String message = "Hello World!"+i;
                    //指定发送消息到哪个路由，以及他的路由键,消息等
                    channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, null, message.getBytes());
                    System.out.println(" [x] Sent '" + message + "'");
                    Thread.sleep(200);
                }
            }catch (Exception e){
                e.printStackTrace();
            }finally {
                //关闭通道和连接
                try {
                    channel.close();
                    connection.close();
                } catch (IOException e) {
                    e.printStackTrace();
                } catch (TimeoutException e) {
                    e.printStackTrace();
                }
            }
        }
    }

消费者现象：这个时候绑定到队列下的所有消费者可进行消费，至于是轮询分发还是公平分发看消费者们的自动应答和一次发送的消息数

当绑定多个队列到路由时，每个队列都能得到一样的消息，再由绑定到队列的消费者消费。例如这个时候再给队QUEUE\_NAME增加一个消费者C1\_1。采用公平分发的规则，那么C1和C1\_1将收到不一样的消息。而QUEUE\_NAME\_SECOND队列下的C2能收到全部的消息

本文重点：路由与队列绑定，以及赋绑定键  
//队列名，路由器名，绑定键

channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,ROUTING_KEY);
    channel.queueBind(QUEUE_NAME_SECOND,EXCHANGE_NAME,ROUTING_KEY);