ITPub博客

首页 > 应用开发 > Java > 每日一学: Routing路由模式

每日一学: Routing路由模式

原创 Java 作者:无敌天骄 时间:2021-04-16 11:44:33 0 删除 编辑

一、模式说明

路由模式特点:

  • 队列与交换机的绑定,不能是任意绑定了,而是要指定一个 RoutingKey (路由key)
  • 消息的发送方在 向 Exchange发送消息时,也必须指定消息的 RoutingKey 。
  • Exchange不再把消息交给每一个绑定的队列,而是根据消息的 Routing Key 进行判断,只有队列的 Routingkey 与消息的 Routing key 完全一致,才会接收到消息











图解:

  • P:生产者,向Exchange发送消息,发送消息时,会指定一个routing key。
  • X:Exchange(交换机),接收生产者的消息,然后把消息递交给 与routing key完全匹配的队列
  • C1:消费者,其所在队列指定了需要routing key 为 error 的消息
  • C2:消费者,其所在队列指定了需要routing key 为 info、error、warning 的消息

二、代码

在编码上与  Publish/Subscribe发布与订阅模式 的区别是交换机的类型为:Direct,还有队列绑定交换机的时候需要指定routing key。

①生产者

package com.itheima.rabbitmq.routing;
import com.itheima.rabbitmq.util.ConnectionUtil; 
import com.rabbitmq.client.BuiltinExchangeType; 
import com.rabbitmq.client.Channel; 
import com.rabbitmq.client.Connection; 
/**
 * 路由模式的交换机类型为:direct 
*/ 
public class Producer { 
    //交换机名称 
    static final String DIRECT_EXCHAGE = "direct_exchange";
     
    //队列名称 
    static final String DIRECT_QUEUE_INSERT = "direct_queue_insert"; 
    
    //队列名称 
    static final String DIRECT_QUEUE_UPDATE = "direct_queue_update"; 
    public static void main(String[] args) throws Exception { 
        //创建连接 
        Connection connection = ConnectionUtil.getConnection(); 
        
        // 创建频道 
        Channel channel = connection.createChannel(); 
        /**
         * 声明交换机 
         * 参数1:交换机名称 
         * 参数2:交换机类型,fanout、topic、direct、headers 
        */ 
        channel.exchangeDeclare(DIRECT_EXCHAGE, BuiltinExchangeType.DIRECT); 
        
        // 声明(创建)队列 
        /**
         * 参数1:队列名称 
         * 参数2:是否定义持久化队列 
         * 参数3:是否独占本次连接 
         * 参数4:是否在不使用的时候自动删除队列 
         * 参数5:队列其它参数 
        */ 
        channel.queueDeclare(DIRECT_QUEUE_INSERT, true, false, false, null); 
        channel.queueDeclare(DIRECT_QUEUE_UPDATE, true, false, false, null); 
        
        //队列绑定交换机 
        channel.queueBind(DIRECT_QUEUE_INSERT, DIRECT_EXCHAGE, "insert"); 
        channel.queueBind(DIRECT_QUEUE_UPDATE, DIRECT_EXCHAGE, "update"); 
        // 发送信息 
        String message = "新增了商品。路由模式;routing key 为 insert " ; 
        
        /**
         * 参数1:交换机名称,如果没有指定则使用默认Default Exchage 
         * 参数2:路由key,简单模式可以传递队列名称 
         * 参数3:消息其它属性 
         * 参数4:消息内容 
        */ 
        channel.basicPublish(DIRECT_EXCHAGE, "insert", null, message.getBytes()); 
        System.out.println("已发送消息:" + message);
        // 发送信息 
        message = "修改了商品。路由模式;routing key 为 update" ; 
        /**
         * 参数1:交换机名称,如果没有指定则使用默认Default Exchage 
         * 参数2:路由key,简单模式可以传递队列名称 
         * 参数3:消息其它属性 
         * 参数4:消息内容 
        */ 
        channel.basicPublish(DIRECT_EXCHAGE, "update", null, message.getBytes()); 
        System.out.println("已发送消息:" + message); 
        // 关闭资源 
        channel.close(); connection.close(); 
    } 
}

②消费者1

package com.itheima.rabbitmq.routing; 
import com.itheima.rabbitmq.util.ConnectionUtil; 
import com.rabbitmq.client.*; 
import java.io.IOException; 
public class Consumer1 { 
    public static void main(String[] args) throws Exception { 
        Connection connection = ConnectionUtil.getConnection(); 
        
        // 创建频道 
        Channel channel = connection.createChannel(); 
        
        //声明交换机 
        channel.exchangeDeclare(Producer.DIRECT_EXCHAGE, BuiltinExchangeType.DIRECT);
         
        // 声明(创建)队列 
        /**
         * 参数1:队列名称 
         * 参数2:是否定义持久化队列 
         * 参数3:是否独占本次连接 
         * 参数4:是否在不使用的时候自动删除队列 
         * 参数5:队列其它参数 
        */ 
        channel.queueDeclare(Producer.DIRECT_QUEUE_INSERT, true, false, false, null); 
        
        //队列绑定交换机 
        channel.queueBind(Producer.DIRECT_QUEUE_INSERT, Producer.DIRECT_EXCHAGE, "insert");
        
        //创建消费者;并设置消息处理 
        DefaultConsumer consumer = new DefaultConsumer(channel){ 
            @Override 
            /**
             * consumerTag 消息者标签,在channel.basicConsume时候可以指定 
             * envelope 消息包的内容,可从中获取消息id,消息routingkey,交换机,消息和重传标志(收到消息失败后是否需要重新发送) 
             * properties 属性信息 
             * body 消息 
            */ 
            public void handleDelivery(String consumerTag, Envelope envelope, 
                    AMQP.BasicProperties properties, byte[] body) throws IOException { 
                //路由key 
                System.out.println("路由key为:" + envelope.getRoutingKey()); 
                
                //交换机 
                System.out.println("交换机为:" + envelope.getExchange()); 
                
                //消息id 
                System.out.println("消息id为:" + envelope.getDeliveryTag()); 
                
                //收到的消息 
                System.out.println("消费者1-接收到的消息为:" + new String(body, "utf-8")); 
            } 
        };
        
        //监听消息 
        /**
         * 参数1:队列名称 
         * 参数2:是否自动确认,设置为true为表示消息接收到自动向mq回复接收到了,mq接收到回复会删除消息,设置为false则需要手动确认 
         * 参数3:消息接收到后回调 
        */ 
        channel.basicConsume(Producer.DIRECT_QUEUE_INSERT, true, consumer); 
    } 
}

③消费者2

package com.itheima.rabbitmq.routing; 
import com.itheima.rabbitmq.util.ConnectionUtil; 
import com.rabbitmq.client.*; 
import java.io.IOException; 
public class Consumer2 { 
    public static void main(String[] args) throws Exception { 
        Connection connection = ConnectionUtil.getConnection(); 
        
        // 创建频道 
        Channel channel = connection.createChannel(); 
        
        //声明交换机 
        channel.exchangeDeclare(Producer.DIRECT_EXCHAGE, BuiltinExchangeType.DIRECT);
        
        // 声明(创建)队列 
        /**
         * 参数1:队列名称 
         * 参数2:是否定义持久化队列 
         * 参数3:是否独占本次连接 
         * 参数4:是否在不使用的时候自动删除队列 
         * 参数5:队列其它参数 
        */ 
        channel.queueDeclare(Producer.DIRECT_QUEUE_UPDATE, true, false, false, null); 
        
        //队列绑定交换机 
        channel.queueBind(Producer.DIRECT_QUEUE_UPDATE, Producer.DIRECT_EXCHAGE, "update");
        
        //创建消费者;并设置消息处理 
        DefaultConsumer consumer = new DefaultConsumer(channel){ 
            @Override 
            /**
             * consumerTag 消息者标签,在channel.basicConsume时候可以指定 
             * envelope 消息包的内容,可从中获取消息id,消息routingkey,交换机,消息和重传标志(收到消息失败后是否需要重新发送) 
             * properties 属性信息 
             * body 消息 
            */ 
            public void handleDelivery(String consumerTag, Envelope envelope, 
                    AMQP.BasicProperties properties, byte[] body) throws IOException { 
                //路由key 
                System.out.println("路由key为:" + envelope.getRoutingKey()); 
                
                //交换机 
                System.out.println("交换机为:" + envelope.getExchange()); 
                
                //消息id 
                System.out.println("消息id为:" + envelope.getDeliveryTag()); 
                
                //收到的消息 
                System.out.println("消费者2-接收到的消息为:" + new String(body, "utf-8")); 
            } 
        };
        //监听消息 
        /**
         * 参数1:队列名称 
         * 参数2:是否自动确认,设置为true为表示消息接收到自动向mq回复接收到了,mq接收到回复会删除消息,设置为false则需要手动确认 
         * 参数3:消息接收到后回调 
        */ 
        channel.basicConsume(Producer.DIRECT_QUEUE_UPDATE, true, consumer); 
    } 
}

三、测试

启动所有消费者,然后使用生产者发送消息;在消费者对应的控制台可以查看到生产者发送对应 routing key 对应队列的消息;到达 按照需要接收的效果。

在执行完测试代码后,其实到RabbitMQ的管理后台找到 Exchanges 选项卡,点击  direct_exchange 的交换机,可以查看到如下的绑定:







四、小结

Routing模式要求队列在绑定交换机时要指定routing key,消息会转发到符合routing key的队列。

喜欢这篇文章的朋友们可以关注个人简介中的公众号

来自 “ ITPUB博客 ” ,链接:https://blog.itpub.net/69964492/viewspace-2768350/ ,如需转载,请注明出处,否则将追究法律责任。

请登录后发表评论 登录
全部评论
公众号"慕容千语" //免费领取面试PDF,架构学习视频

注册时间:2020-02-26

  • 博文量
    79
  • 访问量
    58483