RabbitMQ 之路由和通配符模式，附源码注释讲解

关注
发布于: 2020 年 10 月 31 日
﻿
之前文章分享的是在日常开发过程中的一些基础操作，像RabbitMQ的简单应用、工作队列模式、发布/订阅模式，但是老大交给我的重任，我只写这点怎么可能，所以，我这不是就来了呀，路由模式、通配符模式两种模式也是在日常的开发过程中被使用的，昨天因为篇幅原因没有整理，今天整理出来分享给大家，希望对正在学习的大家有所帮助。
﻿
每一个源码所包含的内容我已经在代码中做好注释了，希望大家看完这两篇文章，五种模式之后能自己去实现一下，不实现你只能是会看，不属于你自己
个人公众号：Java架构师联盟，每日更新技术好文
﻿
好了，RabbitMQ的实战我暂时就整理了这么多，关注我，后期我会更新其他的技术文章，给我点赞+转发支持一波吧，哈哈哈哈
﻿
﻿
路由模式（Routing）消息过滤接收
﻿
﻿
特点：
﻿
生产者发送消息到交换机，同时指定routingKey
一个交换机绑定多个队列，每个队列设置routingKey，并且，一个队列可以设置多个routingKey
交换机通过routingKey来决定把消息转发给哪些队列
其实就是发布订阅模式的加强版。
﻿
二、使用步骤引入库代码如下（示例）：
﻿
<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
</dependency>
﻿
生产者
public class Product {
﻿
    private final static String QUEUE_1 = "queue_1";
    private final static String QUEUE_2 = "queue_2";
    // 交换机名称
    private final static String EXCHANGE_NAME = "exchange_dirct_routing";
​
    // 设置两个routingKey
    private final static String ROUTINGKEY_1 = "info_key";
    private final static String ROUTINGKEY_2 = "error_key";
​
    public static void main(String[] args) throws IOException, TimeoutException {
        // 连接
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("106.**.**.82");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("root");
        factory.setPassword("root");
        Connection connection = null;
        Channel channel = null;
        try{
            connection = factory.newConnection();
            channel = connection.createChannel();
            // 声明交换机
            /**
             * 1. 交换机名称
             * 2、交换机类型:
             *  FANOUT:对应的模式就是 发布/订阅模式
             *  DIRECT:对应 路由(Routing) 的工作模式
             *  TOPIC:对应 Topics 工作模式
             *  HEADERS: 对应 HEADERS 工作模式
             */
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
            
            // 声明队列
            channel.queueDeclare(QUEUE_1,false,false,false, null);
            channel.queueDeclare(QUEUE_2, false, false, false,null);
            // 交换机队列绑定
            /**
             * 1 queue 队列名称
             * 2 exchange 交换机名称
             * 3 routingKey 路由Key
             */
            // 队列(QUEUE_1)绑定routingKey_1
            channel.queueBind(QUEUE_1,EXCHANGE_NAME,ROUTINGKEY_1);
            // 队列(QUEUE_2)绑定routingKey_2和routingKey_1
            channel.queueBind(QUEUE_2,EXCHANGE_NAME,ROUTINGKEY_2);
            channel.queueBind(QUEUE_2,EXCHANGE_NAME,ROUTINGKEY_1);
​
            String message = "这是发送的message biubiubiu~";
​
            // 参数：String exchange, String routingKey, BasicProperties props, byte[] body
            /**
             * exchange:交换机 如果不指定("")，就默认交换机
             * routingKey：路由key；交换机根据路由key将消息转发到指定的队列，如果使用默认交换机，routingKey为队列名称
             * props:额外属性
             * body：消息内容
             */
            for(int i=0;i<4;i++){
                // 修改这里的routingkey 你可以看到不同的效果喔
                // channel.basicPublish(EXCHANGE_NAME,ROUTINGKEY_1,null,message.getBytes());
                channel.basicPublish(EXCHANGE_NAME,ROUTINGKEY_2,null,message.getBytes());
                System.out.println("发送了message："+message+", routingKey: "+ROUTINGKEY_2);
            }
        } catch (TimeoutException | IOException e) {
            e.printStackTrace();
        } finally {
            channel.close();
            connection.close();
        }
    }
}
﻿
消费者1
/**
 * 对应QUEUE_1的消费者
 */
public class Consumer_1 {
​
    private final static String QUEUE_1 = "queue_1";
    private final static String QUEUE_2 = "queue_2";
    private final static String EXCHANGE_NAME = "exchange_dirct_routing";
​
    private final static String ROUTINGKEY_1 = "info_key";
    private final static String ROUTINGKEY_2 = "error_key";
​
    public static void main(String[] args) {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("106.**.**.82");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("root");
        factory.setPassword("root");
​
        Connection connection = null;
        Channel channel = null;
        try {
            connection = factory.newConnection();
            channel = connection.createChannel();
            // 声明交换机
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
            // 声明队列
            channel.queueDeclare(QUEUE_1,false,false,false, null);
//            channel.queueDeclare(QUEUE_2, false, false, false,null);
​
            // 交换机队列绑定
            channel.queueBind(QUEUE_1,EXCHANGE_NAME,ROUTINGKEY_1);
​
//            channel.queueBind(QUEUE_2,EXCHANGE_NAME,ROUTINGKEY_2);
//            channel.queueBind(QUEUE_2,EXCHANGE_NAME,ROUTINGKEY_1);
​
            DefaultConsumer consumer = new DefaultConsumer(channel){
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
                    String message = new String (body, StandardCharsets.UTF_8);
                    System.out.println("receive:" + message);
                }
            };
​
            // 接收消息 监听队列
            // 参数：String queue, boolean autoAck, Consumer callback
            /**
             * queue:队列
             * autoAck:自动回复：当消费者接收到消息后，告诉mq消息已经接收。TRUE：自动回复，false：编程回复
             * callback:消费方法，当消费者接收消息执行的方法。
             */
            channel.basicConsume(QUEUE_1,true,consumer);
​
        }catch (Exception e){
            e.printStackTrace();
        }
    }
}
﻿
消费者2
public class Consumer_2 {
    private final static String QUEUE_1 = "queue_1";
    private final static String QUEUE_2 = "queue_2";
    private final static String EXCHANGE_NAME = "exchange_dirct_routing";
​
    private final static String ROUTINGKEY_1 = "info_key";
    private final static String ROUTINGKEY_2 = "error_key";
​
    public static void main(String[] args) {
​
        ConnectionFactory factory = new ConnectionFactory();
​
        factory.setHost("106.**.**.82");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("root");
        factory.setPassword("root");
​
        Connection connection = null;
        Channel channel = null;
        try {
            connection = factory.newConnection();
            channel = connection.createChannel();
            // 声明交换机
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
            // 声明队列
            channel.queueDeclare(QUEUE_2, false, false, false,null);
​
            // 交换机队列绑定
            /**
             * 1 queue 队列名称
             * 2 exchange 交换机名称
             * 3 routingKey 路由Key
             */
            channel.queueBind(QUEUE_2,EXCHANGE_NAME,ROUTINGKEY_2);
            channel.queueBind(QUEUE_2,EXCHANGE_NAME,ROUTINGKEY_1);
​
            DefaultConsumer consumer = new DefaultConsumer(channel){
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
                    String message = new String (body, StandardCharsets.UTF_8);
                    System.out.println("receive:" + message);
                }
            };
​
            // 接收消息 监听队列
            // 参数：String queue, boolean autoAck, Consumer callback
            /**
             * queue:队列
             * autoAck:自动回复：当消费者接收到消息后，告诉mq消息已经接收。TRUE：自动回复，false：编程回复
             * callback:消费方法，当消费者接收消息执行的方法。
             */
            channel.basicConsume(QUEUE_2,true,consumer);
​
        }catch (Exception ignored){
            ignored.printStackTrace();
        }
    }
​
}
﻿
通配符模式(Topic)路由模式，routingKey采用通配符模式。
﻿
﻿
特点：
﻿
跟routing模式相比，Topics模式是通配符匹配
﻿
* : 只能匹配一个词，例如: info.*，他可以匹配 info.hello，info.world
# : 可以匹配多个词，例如: info.#，它可以匹配 info.hello.world，info.xx，info
# 可以匹配空
﻿
二、使用步骤引入库
<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
</dependency>
﻿
生产者两个队列 支付宝 和 微信， 有些人想要支付宝，有些人想要微信 ，而有些人全都要
﻿
public class TopicProduct {
    // 两个队列 支付宝 和 微信， 有些人想要支付宝，有些人想要微信 有些人全都要
    private final static String QUEUE_1 = "queue_topic_alipay";
    private final static String QUEUE_2 = "queue_topic_wechat";
    private final static String EXCHANGE_NAME = "exchange_topic";
​
    /**
     * 可以匹配 info.alipay,info.alipay.wechat...
     * 当 info.alipay.wechat 可以同时匹配两个路由key
     */
    private final static String ROUTINGKEY_1 = "info.#.alipay.#";
    private final static String ROUTINGKEY_2 = "info.#.wechat.#";
​
    public static void main(String[] args) throws IOException, TimeoutException {
        // 连接
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("106.**.**.82");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("root");
        factory.setPassword("root");
        Connection connection = null;
        Channel channel = null;
        try{
            connection = factory.newConnection();
            channel = connection.createChannel();
            // 声明交换机
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
            // 声明队列
            channel.queueDeclare(QUEUE_1,false,false,false,null);
            channel.queueDeclare(QUEUE_2,false,false,false,null);
​
            /* 交换机队列绑定
             * 1 queue 队列名称
             * 2 exchange 交换机名称
             * 3 routingKey 路由Key
             */
            channel.queueBind(QUEUE_1,EXCHANGE_NAME,ROUTINGKEY_1);
            channel.queueBind(QUEUE_2,EXCHANGE_NAME,ROUTINGKEY_2);
​
            String message = "";
​
            // 参数：String exchange, String routingKey, BasicProperties props, byte[] body
            /**
             * exchange:交换机 如果不指定("")，就默认交换机
             * routingKey：路由key；交换机根据路由key将消息转发到指定的队列，如果使用默认交换机，routingKey为队列名称
             * props:额外属性
             * body：消息内容
             */
            message = "这个是给喜欢 AliPay 的朋友们的~";
            channel.basicPublish(EXCHANGE_NAME,"info.alipay",null,message.getBytes());
            System.out.println(new Date()+":发送了message："+message+", Topic routingKey: "+ROUTINGKEY_1);
​
            message = "喜欢 wechat 的朋友，接好了";
            channel.basicPublish(EXCHANGE_NAME,"info.wechat",null,message.getBytes());
            System.out.println(new Date()+":发送了："+message+", Topic routingKey:"+ROUTINGKEY_2);
​
            message = "同时喜欢 ALIPAY 和 WECHAT 的朋友接好了";
            channel.basicPublish(EXCHANGE_NAME,"info.alipay.wechat",null,message.getBytes());
            System.out.println(new Date()+":发送了："+message+",Topic routingKeys:"+ROUTINGKEY_2+","+ROUTINGKEY_1);
​
        } catch (TimeoutException | IOException e) {
            e.printStackTrace();
        } finally {
            channel.close();
            connection.close();
        }
    }
}
﻿
消费者1喜欢alipay的
﻿
public class Consumer_Topic_1 {
    private final static String QUEUE_1 = "queue_topic_alipay";
    //private final static String QUEUE_2 = "queue_topic_wechat";
    private final static String EXCHANGE_NAME = "exchange_topic";
    private final static String ROUTINGKEY_1 = "info.#.alipay.#";
    //private final static String ROUTINGKEY_2 = "info.#.wechat.#";
​
    public static void main(String[] args) {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("106.**.**.82");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("root");
        factory.setPassword("root");
​
        Connection connection = null;
        Channel channel = null;
        try {
            connection = factory.newConnection();
            channel = connection.createChannel();
            // 声明交换机
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
            // 声明队列
            channel.queueDeclare(QUEUE_1,false,false,false, null);
​
            // 交换机队列绑定
            channel.queueBind(QUEUE_1,EXCHANGE_NAME,ROUTINGKEY_1);
​
            DefaultConsumer consumer = new DefaultConsumer(channel){
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
                    String message = new String (body, StandardCharsets.UTF_8);
                    System.out.println("receive:" + message);
                }
            };
​
            // 接收消息 监听队列
            // 参数：String queue, boolean autoAck, Consumer callback
            /**
             * queue:队列
             * autoAck:自动回复：当消费者接收到消息后，告诉mq消息已经接收。TRUE：自动回复，false：编程回复
             * callback:消费方法，当消费者接收消息执行的方法。
             */
            channel.basicConsume(QUEUE_1,true,consumer);
​
        }catch (Exception e){
            e.printStackTrace();
        }
    }
}
​
﻿
输出如下：
﻿
receive:这个是给喜欢 AliPay 的朋友们的~
receive:喜欢 ALIPAY 和 WECHAT 接好了
﻿
消费者2喜欢wechat
﻿
public class Consumer_Topic_2 {
    // private final static String QUEUE_1 = "queue_topic_alipay";
    private final static String QUEUE_2 = "queue_topic_wechat";
    private final static String EXCHANGE_NAME = "exchange_topic";
​
    // private final static String ROUTINGKEY_1 = "info.#.alipay.#";
    private final static String ROUTINGKEY_2 = "info.#.wechat.#";
​
    public static void main(String[] args) {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("106.**.**.82");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("root");
        factory.setPassword("root");
​
        Connection connection = null;
        Channel channel = null;
        try {
            connection = factory.newConnection();
            channel = connection.createChannel();
            // 声明交换机
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
            // 声明队列
            channel.queueDeclare(QUEUE_2, false, false, false,null);
​
            // 交换机队列绑定
            /**
             * 1 queue 队列名称
             * 2 exchange 交换机名称
             * 3 routingKey 路由Key
             */
            channel.queueBind(QUEUE_2,EXCHANGE_NAME,ROUTINGKEY_2);
​
            DefaultConsumer consumer = new DefaultConsumer(channel){
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
                    String message = new String (body, StandardCharsets.UTF_8);
                    System.out.println("receive:" + message);
                }
            };
​
            // 接收消息 监听队列
            // 参数：String queue, boolean autoAck, Consumer callback
            /**
             * queue:队列
             * autoAck:自动回复：当消费者接收到消息后，告诉mq消息已经接收。TRUE：自动回复，false：编程回复
             * callback:消费方法，当消费者接收消息执行的方法。
             */
            channel.basicConsume(QUEUE_2,true,consumer);
​
        }catch (Exception ignored){
            ignored.printStackTrace();
        }
    }
}
​
​
﻿
输出如下：
﻿
receive:喜欢 wechat 的朋友，接好了
receive:喜欢 ALIPAY 和 WECHAT 接好了
﻿
发布于: 2020 年 10 月 31 日阅读数: 161
原文链接:【http://xie.infoq.cn/article/f299eacddc1d748494a0b8dfd】。
小Q

关注
还未添加个人签名 2020.06.30 加入
小Q 公众号：Java架构师联盟作者多年从事一线互联网Java开发的学习历程技术汇总，旨在为大家提供一个清晰详细的学习教程，侧重点更倾向编写Java核心内容。如果能为您提供帮助，请给予支持(关注、点赞、分享)！
发布
暂无评论
创作场景

RabbitMQ 之路由和通配符模式，附源码注释讲解

路由模式（Routing）

二、使用步骤

引入库

生产者

消费者1

消费者2

通配符模式(Topic)

二、使用步骤

引入库

生产者

消费者1

消费者2

小Q

评论
