RabbitMQ-路由模式routing

刺骨的言语ヽ痛彻心扉 2022-05-30 10:53 269阅读 0赞

### 文章目录 ###

*  RabbitMQ-路由模式routing
 *   *   *   *  1.Exchange
             *   *  生产者
                 *  消费者1
                 *  消费者2

# RabbitMQ-路由模式routing #

#### 1.Exchange ####

>  交换机，转发器，一方面接收生产者的消息，另一方面向队列推送消息

>  匿名转发(第一个参数为"")

channel.basicPublish("","",null,msg.getBytes());

>  fanout(不处理路由键)：每个和交换机绑定的队列都会收到消息

channel.exchangeDeclare(EXCHANGER_NAME,"fanout");
    channel.basicPublish(EXCHANGER_NAME,"",null,msg.getBytes());

![这里写图片描述][70]

>  direct(处理路由键)：会根据发送的key键发送给对应队列

![这里写图片描述][70 1]

>  路由模式：例如发送消息带有error key，则两个队列都会收到消息，若只发送info key，下方队列会收到消息

![这里写图片描述][70 2]

##### 生产者 #####

package com.ithzk.rabbitmq.routing;
    
    import com.ithzk.rabbitmq.utils.RabbitMQConnectionUtils;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /** * @author hzk * @date 2018/3/10 */
    public class Send { 
    
        private final static String EXCHANGER_NAME = "test_exchange_direct";
    
        public static void main(String[] args) throws IOException, TimeoutException { 
    
            Connection connection = RabbitMQConnectionUtils.getConnection();
    
            Channel channel = connection.createChannel();
    
            // 声明交换机
            channel.exchangeDeclare(EXCHANGER_NAME,"direct");
    
            String msg = "hello exchange direct";
    
            //发送error两个队列都会收到消息
            String routingKey = "error";
            //发送info 只有消费者2会收到消息
            String routingKey2 = "info";
            channel.basicPublish(EXCHANGER_NAME,routingKey2,null,msg.getBytes());
    
            System.out.println("Send msg"+msg);
    
            channel.close();
            connection.close();
        }
    
    }

##### 消费者1 #####

package com.ithzk.rabbitmq.routing;
    
    import com.ithzk.rabbitmq.utils.RabbitMQConnectionUtils;
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /** * @author hzk * @date 2018/3/10 */
    public class Recv1 { 
    
        private static final String QUEUE_NAME="test_queue_direct_one";
        private final static String EXCHANGER_NAME = "test_exchange_direct";
    
        public static void main(String[] args) throws IOException, TimeoutException { 
            //获取连接
            Connection connection = RabbitMQConnectionUtils.getConnection();
    
            //从连接中获取频道
            final Channel channel = connection.createChannel();
    
            //声明队列
            channel.queueDeclare(QUEUE_NAME,false,false,false,null);
    
            String routingKey = "error";
            //绑定队列到交换机 转发器
            channel.queueBind(QUEUE_NAME,EXCHANGER_NAME,routingKey);
    
            //保证一次只发一个
            channel.basicQos(1);
    
            DefaultConsumer consumer = new DefaultConsumer(channel) { 
    
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                    String msg = new String(body, "utf-8");
                    System.out.println("[1] Recv msg:" + msg);
    
                    try { 
                        Thread.sleep(10000);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    } finally { 
                        System.out.println("[1] done");
                        channel.basicAck(envelope.getDeliveryTag(), false);
                    }
                }
            };
    
           boolean autoAck = false;
    
           channel.basicConsume(QUEUE_NAME,autoAck,consumer);
    
           System.out.println("[Consumer 1 start]");
    
    
        }
    
    
    }

##### 消费者2 #####

package com.ithzk.rabbitmq.routing;
    
    import com.ithzk.rabbitmq.utils.RabbitMQConnectionUtils;
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /** * @author hzk * @date 2018/3/10 */
    public class Recv2 { 
    
        private static final String QUEUE_NAME="test_queue_direct_two";
        private final static String EXCHANGER_NAME = "test_exchange_direct";
    
        public static void main(String[] args) throws IOException, TimeoutException { 
            //获取连接
            Connection connection = RabbitMQConnectionUtils.getConnection();
    
            //从连接中获取频道
            final Channel channel = connection.createChannel();
    
            //声明队列
            channel.queueDeclare(QUEUE_NAME,false,false,false,null);
    
            //绑定队列到交换机 转发器
            String routingKey = "error";
            String routingKey2 = "info";
            String routingKey3 = "warning";
            channel.queueBind(QUEUE_NAME,EXCHANGER_NAME,routingKey);
            channel.queueBind(QUEUE_NAME,EXCHANGER_NAME,routingKey2);
            channel.queueBind(QUEUE_NAME,EXCHANGER_NAME,routingKey3);
    
            //保证一次只发一个
            channel.basicQos(1);
    
            DefaultConsumer consumer = new DefaultConsumer(channel) { 
    
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                    String msg = new String(body, "utf-8");
                    System.out.println("[2] Recv msg:" + msg);
    
                    try { 
                        Thread.sleep(10000);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    } finally { 
                        System.out.println("[2] done");
                        channel.basicAck(envelope.getDeliveryTag(), false);
                    }
                }
            };
    
           boolean autoAck = false;
    
           channel.basicConsume(QUEUE_NAME,autoAck,consumer);
    
           System.out.println("[Consumer 2 start]");
    
    
        }
    
    
    }

![这里写图片描述][70 3]

[70]: /images/20220530/b6105793db7b499a8ab96f38c2c6a9e7.png
[70 1]: /images/20220530/313e066b64c14f4fa7ce3eb7fb7316b3.png
[70 2]: /images/20220530/b3e7b50cf28f4bb78f2b6b604c06c1dd.png
[70 3]: /images/20220530/cc50ad376e184d5abf527c124f4c1505.png