rabbitmq 7种队列实现java版

深碍√TFBOYSˉ_ 2023-07-06 14:44 22阅读 0赞

### 文章目录 ###

*  rabbitmq7种实现方式
 *  搭建maven项目
 *   *  引入依赖
     *  创建连接
 *  简单队列
 *   *  消息生产者
     *  消息消费者
 *  work queues 工作队列
 *   *  生产者
     *  消费者
     *  能者多劳（公平分发）：消费能力强则消费更多消息
 *  Publish/Subscribe 发布订阅模式
 *   *  生产者代码
     *  消费者
     *  控制台观察该交换器
     *  交换机
 *  Routing 路由模式
 *  topic 主题模式（通配符匹配模式）

# rabbitmq7种实现方式 #

在rabbitmq的官网 https://www.rabbitmq.com/getstarted.html 有给出7中实现方式：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70]  
![在这里插入图片描述][20200219224922976.png]  
这里依次学习。

# 搭建maven项目 #

## 引入依赖 ##

<dependency>
        <groupId>com.rabbitmq</groupId>
        <artifactId>amqp-client</artifactId>
        <version>5.8.0</version>
    </dependency>

## 创建连接 ##

public static Connection getConnection() throws Exception { 
        //定义连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        //设置服务地址
        factory.setHost("localhost");
        //端口
        factory.setPort(5672);
        //设置账号信息，用户名、密码、vhost
        factory.setVirtualHost("/admin_vhost");
        factory.setUsername("admin");
        factory.setPassword("123456");
        // 通过工程获取连接
        Connection connection = factory.newConnection();
        return connection;
    }

# 简单队列 #

![在这里插入图片描述][20200219204124553.png]  
**顾名思义，最简单的队列，有生产者和消费者两个角色，生产者生产消息，将消息发送到队列汇总，消费者从队列中消费消息。**

## 消息生产者 ##

下面定义一个队列名称为`q_test_01`的队列，并且向其发送一条消息。

private final static String QUEUE_NAME = "q_test_01";
    
    publicstatic void main(String[] argv) throws Exception { 
        // 获取到连接以及mq通道
        Connection connection = ConnectionUtil.getConnection();
        // 从连接中创建通道
        Channel channel = connection.createChannel();
    
        // 声明（创建）队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
        // 消息内容
        String message = "Hello World!";
        channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
        System.out.println(" [x] Sent '" + message + "'");
        //关闭通道和连接
        channel.close();
        connection.close();
    }

执行上面main方法，发送一条消息。  
在控制台可以看到有一条消息发送到了rabbitmq队列中：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 1]  
在queue标签中可以看到新建的队列  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 2]  
点击上面的`q_test_01`链接进入队列详情页面，里面有`Get messages`功能，可以获取队列中的消息，可以看到消息内容`Hello World!` :  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 3]

## 消息消费者 ##

消费者消费队列中的数据：

private final static String QUEUE_NAME = "q_test_01";
    
    public static void main(String[] argv) throws Exception { 
    
        // 获取到连接以及mq通道
        Connection connection = ConnectionUtil.getConnection();
        // 从连接中创建通道
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
        //回调函数。队列中有数据后会回调该方法
        Consumer consumer = new DefaultConsumer(channel) { 
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                                       byte[] body) throws IOException { 
                String message = new String(body, "UTF-8");
                System.out.println(" [x] Received '" + message);
            }
        };
    
        //channel监听队列。
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }

上面采用了监听回调机制来消费数据。当消费者监听到有数据以后会回调Consumer 的handleDelivery方法。

注意channel.basicConsume的第二个参数我这里传的true，表示自动提交ack响应。在一些业务场景中可能会要求关闭自动提交，而是确定消费成功后才手动提交ack。

启动消费者方法，可以看到打印输出消息：  
![在这里插入图片描述][20200219210141801.png]  
消费者不会停止服务，会一直监听。此时生产者再发送一条消息，消费者也可以马上收到消息并打印：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 4]

# work queues 工作队列 #

多消费这模式  
![在这里插入图片描述][20200219210518237.png]

## 生产者 ##

针对上面生产者做一点修改，连续发送多条数据。主要是加了一个for循环，连续发送50条消息。

private final static String QUEUE_NAME = "q_test_workqueue";
    
    public static void main(String[] argv) throws Exception { 
        // 获取到连接以及mq通道
        Connection connection = ConnectionUtil.getConnection();
        // 从连接中创建通道
        Channel channel = connection.createChannel();
    
        // 声明（创建）队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
    
    
        // 消息内容
        for(int i = 0 ; i < 50 ; i ++){ 
            String message = "Hello rabbitmq! " + i;
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            System.out.println(" [x] Sent '" + message + "'");
    
            TimeUnit.MILLISECONDS.sleep(100);
        }
    
    
        //关闭通道和连接
        channel.close();
        connection.close();
    }

## 消费者 ##

这里定义两个消费者，一个消费者每条消息耗时1秒，另一个性能好，每条消息耗时2秒。使用sleep来模拟耗时。

消费者1：

public class Recv1 { 
    
        private final static String QUEUE_NAME = "q_test_workqueue";
    
        public static void main(String[] argv) throws Exception { 
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            // 从连接中创建通道
            Channel channel = connection.createChannel();
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            //回调函数。队列中有数据后会回调该方法
            Consumer consumer = new DefaultConsumer(channel) { 
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                                           byte[] body) throws IOException { 
                    String message = new String(body, "UTF-8");
                    System.out.println(" [1] Received : " + message);
                    try { 
                        TimeUnit.SECONDS.sleep(1);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                    System.out.println(" [1] done : " + message);
                }
            };
    
            //channel监听队列。
            channel.basicConsume(QUEUE_NAME, true, consumer);
        }
    }

消费者2：

public class Recv2 { 
    
        private final static String QUEUE_NAME = "q_test_workqueue";
    
        public static void main(String[] argv) throws Exception { 
    
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            // 从连接中创建通道
            Channel channel = connection.createChannel();
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            //回调函数。队列中有数据后会回调该方法
            Consumer consumer = new DefaultConsumer(channel) { 
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                                           byte[] body) throws IOException { 
                    String message = new String(body, "UTF-8");
                    System.out.println(" [2] Received : " + message);
                    try { 
                        TimeUnit.SECONDS.sleep(2);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                    System.out.println(" [2] done : " + message);
                }
            };
    
            //channel监听队列。
            channel.basicConsume(QUEUE_NAME, true, consumer);
        }
    }

首先启动两个消费者，使其准备好消费消息。然后启动生产者连续发送50条消息。从控制台可以看到两个消费者消费情况  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 5]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 6]  
通过上述实际结果，可以看到，两个消费者消费的消息总数量是相同的，消息不会被不同消费者重复消费。

由于我们设置了两个消费者消费能力（sleep时间）不同，按理来说应该消费者1消费的多。

之所以两个消费者消费消息数量相同是因为rabbitmq默认采用`轮训策略`（round-robin）进行分配消息。

## 能者多劳（公平分发）：消费能力强则消费更多消息 ##

要想实现公平分发，必须关闭自动提交ack，改为手动ack。就是`channel.basicConsume(QUEUE_NAME, true, consumer);`的第二个参数。  
官方解释：

autoAck true if the server should consider messages acknowledged once delivered; 
    false if the server should expect explicit acknowledgements

另外，很重要的 一点是，生产者和消费者都要配置一下`basicQos`，标明收到消费者相应ack前可以发送多少条消息。默认是0，不限制。

/** * Request a specific prefetchCount "quality of service" settings * for this channel. * * @see #basicQos(int, int, boolean) * @param prefetchCount maximum number of messages that the server * will deliver, 0 if unlimited * @throws java.io.IOException if an error is encountered */
    void basicQos(int prefetchCount) throws IOException;

我们这里设置为`channel.basicQos(1);`保证消费一条发一条给消费者。

注意这里不是只生产者发送到队列，而是说的从队列中发送到消费者的消息数量。

接下来,由于关闭了自动ack ,还需要在消费者Consumer的finally中手动提交ack。  
完整的生产者代码：

public class Send { 
    
        private final static String QUEUE_NAME = "q_test_workqueue";
    
        public static void main(String[] argv) throws Exception { 
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            // 从连接中创建通道
            Channel channel = connection.createChannel();
    
            // 声明（创建）队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            //在收到消费者响应之前，最多发送多少条消息给消费者
            //用于能者多劳模式
            channel.basicQos(1);
    
            // 消息内容
            for(int i = 0 ; i < 50 ; i ++) { 
                String message = "Hello rabbitmq! " + i;
                channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
                System.out.println(" [x] Sent '" + message + "'");
    
                TimeUnit.MILLISECONDS.sleep(100);
            }
    
            //关闭通道和连接
            channel.close();
            connection.close();
        }
    }

完整的消费者1：

public class Recv1 { 
    
        private final static String QUEUE_NAME = "q_test_workqueue";
    
        public static void main(String[] argv) throws Exception { 
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            // 从连接中创建通道
            Channel channel = connection.createChannel();
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            channel.basicQos(1);
    
            //回调函数。队列中有数据后会回调该方法
            Consumer consumer = new DefaultConsumer(channel) { 
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                                           byte[] body) throws IOException { 
                    String message = null;
    
                    try { 
                        message = new String(body, "UTF-8");
                        System.out.println(" [1] Received : " + message);
    
                        TimeUnit.SECONDS.sleep(1);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    } finally { 
                        System.out.println(" [1] done : " + message);
                        //手动提交ack回执
                        channel.basicAck(envelope.getDeliveryTag(), false);
                    }
                }
            };
    
            //channel监听队列。
            channel.basicConsume(QUEUE_NAME, false, consumer);
        }
    }

完整消费者2代码

public class Recv2 { 
    
        private final static String QUEUE_NAME = "q_test_workqueue";
    
        public static void main(String[] argv) throws Exception { 
    
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            // 从连接中创建通道
            Channel channel = connection.createChannel();
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            channel.basicQos(1);
    
            //回调函数。队列中有数据后会回调该方法
            Consumer consumer = new DefaultConsumer(channel) { 
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                                           byte[] body) throws IOException { 
                    String message = null;
                    try { 
                        message = new String(body, "UTF-8");
                        System.out.println(" [2] Received : " + message);
    
                        TimeUnit.SECONDS.sleep(2);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    } finally { 
                        System.out.println(" [2] done : " + message);
                        //手动提交ack回执
                        channel.basicAck(envelope.getDeliveryTag(), false);
                    }
                }
            };
    
            //channel监听队列。
            channel.basicConsume(QUEUE_NAME, false, consumer);
        }
    }

首先运行消费者1和2，然后启动生产者。控制台输出：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 7]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 8]  
可以看到，消费者1消费的消息数量是消费者2的两倍，因为消费者1休眠时间是消费者2的一半。

# Publish/Subscribe 发布订阅模式 #

![在这里插入图片描述][20200219221036411.png]

上图中X指交换机exchange。  
发布订阅模式有以下特点：

*  一个生产者多个消费者
 *  每一个消费者都有自己的一个队列
 *  生产者没有将消息直接发送到队列，而是发送到了交换机
 *  每个队列都要绑定到交换机
 *  生产者发送的消息，经过交换机，到达队列，实现，一个消息被多个消费者获取的目的  
    注意：一个消费者队列可以有多个消费者实例，只有其中一个消费者实例会消费

以上特性类似于Kafka的topic和partition，一个topic的消息发送到不同的partition上，一个partition可以被一个消费者组内的一个消费者消费。更多Kafka知识请参考Kafka官网。

**下面模拟场景为：一个生产者，通过交换机发送10条消息。交换机采用的是`BuiltinExchangeType.FANOUT`类型进行交换转发。生产者绑定两个队列。定义三个消费者，消费者1消费队列1，消费者2和3消费队列2**

## 生产者代码 ##

生产者跟之前相比，多了以下几步：

*  定义交换机
 *  将队列绑定到交换机上

完整代码：

public class Send { 
    
        private final static String QUEUE_NAME1 = "q_test_publish_subscribe1";
        private final static String QUEUE_NAME2 = "q_test_publish_subscribe2";
        private final static String EXCHANGE_NAME = "q_test_publish_subscribe_exchange";
    
        public static void main(String[] argv) throws Exception { 
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            // 从连接中创建通道
            Channel channel = connection.createChannel();
    
            //定义交换机
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
    
            // 声明（创建）队列. durable=true表示该队列需要持久化
            channel.queueDeclare(QUEUE_NAME1, true, false, false, null);
            channel.queueDeclare(QUEUE_NAME2, true, false, false, null);
    
            //将队列绑定到交换机上
            channel.queueBind(QUEUE_NAME1, EXCHANGE_NAME, "");
            channel.queueBind(QUEUE_NAME2, EXCHANGE_NAME, "");
    
            //在收到消费者响应之前，最多发送多少条消息给消费者
            //用于能者多劳模式
            channel.basicQos(1);
    
            // 消息内容
            for (int i = 0; i < 10; i++) { 
                String message = "Hello rabbitmq! " + i;
                channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
                System.out.println(" [x] Sent '" + message + "'");
    
                TimeUnit.MILLISECONDS.sleep(100);
            }
    
            //关闭通道和连接
            channel.close();
            connection.close();
        }
    }

## 消费者 ##

消费者有3个，唯一不同之处就是消费的topic不同。

消费者1：

public class Recv1 { 
    
        private final static String QUEUE_NAME = "q_test_publish_subscribe1";
    
        public static void main(String[] argv) throws Exception { 
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            // 从连接中创建通道
            Channel channel = connection.createChannel();
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, true, false, false, null);
    
            channel.basicQos(1);
    
            //回调函数。队列中有数据后会回调该方法
            Consumer consumer = new DefaultConsumer(channel) { 
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                                           byte[] body) throws IOException { 
                    String message = null;
    
                    try { 
                        message = new String(body, "UTF-8");
                        System.out.println(" [1] Received : " + message);
                    } finally { 
                        System.out.println(" [1] done : " + message);
                        //手动提交ack回执
                        channel.basicAck(envelope.getDeliveryTag(), false);
                    }
                }
            };
    
            //channel监听队列。
            channel.basicConsume(QUEUE_NAME, false, consumer);
        }
    }

消费者2和消费者3消费的topic是`q_test_publish_subscribe2`，其他与消费者1完全相同。

首先启动消费者1，2，3；然后启动生产者发送消息。  
控制台输出:  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 9]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 10]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 11]

> 可以看到，两个队列的数据是相同的，也就是说`BuiltinExchangeType.FANOUT`类型的交换机将消息复制到了每个队列中。消费者1单独消费q\_test\_publish\_subscribe1队列中的消息，所以有10条记录。q\_test\_publish\_subscribe2队列由Recv2和Recv3共同消费，由于他俩这里消费速度相同，所以各消费了5条数据。

## 控制台观察该交换器 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 12]  
点击进入交换机详情页：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 13]  
可以看到有两个队列绑定在这里。而且控制台支持手动解绑。  
在队列页面也可以看到这两个队列：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 14]

## 交换机 ##

交换机的作用是接收生产者的消息，并将其根据一定规则发送到队列中。注意交换机本身不保存消息，rabbitmq中消息都是保存在队列中。

交换机支持以下4中模式：  
![在这里插入图片描述][20200219225416754.png]

*  `fanout`：fanout类似于广播，会将受到的消息发送给每一个绑定的队列中。
 *  `direct`：路由模式
 *  `topic`：
 *  `headers`：

# Routing 路由模式 #

与发布订阅相比，路由模式最主要的修改点就是交换机类型采用`BuiltinExchangeType.DIRECT`。

根据官网实例，模拟一个生产者，一个`BuiltinExchangeType.DIRECT`类型的交换机，两个队列，两个消费者。  
error类型的消息同时发送到两个队列。info和warning类型的消息只发送到一个队列。  
![在这里插入图片描述][20200219230701966.png]  
生产者完整代码：

public class Send { 
        private final static String QUEUE_NAME1 = "q_test_routing1";
        private final static String QUEUE_NAME2 = "q_test_routing2";
        private final static String EXCHANGE_NAME = "q_test_routing_exchange";
    
        private final static String ERROR = "error";
        private final static String INFO = "info";
        private final static String WARNING = "warning";
    
        public static void main(String[] argv) throws Exception { 
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            // 从连接中创建通道
            Channel channel = connection.createChannel();
    
            //定义交换机
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
    
            // 声明（创建）队列. durable=true表示该队列需要持久化
            channel.queueDeclare(QUEUE_NAME1, true, false, false, null);
            channel.queueDeclare(QUEUE_NAME2, true, false, false, null);
    
            //将队列绑定到交换机上
            channel.queueBind(QUEUE_NAME1, EXCHANGE_NAME, ERROR);
            channel.queueBind(QUEUE_NAME2, EXCHANGE_NAME, ERROR);
            channel.queueBind(QUEUE_NAME2, EXCHANGE_NAME, INFO);
            channel.queueBind(QUEUE_NAME2, EXCHANGE_NAME, WARNING);
    
            //在收到消费者响应之前，最多发送多少条消息给消费者
            //用于能者多劳模式
            channel.basicQos(1);
    
            // 消息内容. 这里error，info，warning的消息各发送2个
            channel.basicPublish(EXCHANGE_NAME, ERROR, null, "Hello rabbitmq! error1".getBytes());
            channel.basicPublish(EXCHANGE_NAME, ERROR, null, "Hello rabbitmq! error2".getBytes());
            channel.basicPublish(EXCHANGE_NAME, INFO, null, "Hello rabbitmq! info1".getBytes());
            channel.basicPublish(EXCHANGE_NAME, INFO, null, "Hello rabbitmq! info2".getBytes());
            channel.basicPublish(EXCHANGE_NAME, WARNING, null, "Hello rabbitmq! warning1".getBytes());
            channel.basicPublish(EXCHANGE_NAME, WARNING, null, "Hello rabbitmq! warning2".getBytes());
    
            //关闭通道和连接
            channel.close();
            connection.close();
        }
    }

消费者1和消费者2的代码拷贝自发布订阅模式的消费者代码，唯一不同之处就是队列名称，这里就不贴完整代码了。

首先运行两个消费者，然后运行生产者。控制台输出如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 15]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 16]  
与预期一致，error类型的消息同时发送到两个队列，两个消费者都消费到了error的消息。info和warning类型的消息只发送到一个队列，只有消费者2消费到。

# topic 主题模式（通配符匹配模式） #

总感觉rabbitmq的命名与常见定义不符。  
这里topic可以理解为通配符模式。消息发向exchange交换器，然后根据不同的通配符进行匹配转发。  
通配符有两种：

*  \*：只能向后或向前匹配一个词
 *  \#：可以匹配0个或多个

官网介绍：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 17]

接下来写个demo实例。

说明：之前队列绑定都是在生产者做的，很多时候在消费者进行绑定会更好，因为消费者可以根据自己的需求灵活消费想要消费的数据。

消息生产者定义了那么几个routingKey，消息将根据这些routingKey进行路由：  
![在这里插入图片描述][2020022113504116.png]  
定义三个消费者，每个消费者都有自己的队列。

*  `消费者1`消费routingKey为`orderlog.*`的消息；它可以匹配`orderlog.error`和`orderlog.info`两个routingKey的消息
 *  `消费者2`消费routingKey为`orderlog.error`的消息；它只能匹配到`orderlog.error`的消息
 *  `消费者3`消费routingKey为`orderlog.#`的消息；它可以匹配`orderlog.error`，`orderlog.info.xx`和`orderlog.info`三个routingKey的消息

下面完整代码：

public class Send { 
    
        private final static String EXCHANGE_NAME = "q_test_topic_exchange";
        private final static String ERROR = "orderlog.error";
        private final static String INFO = "orderlog.info";
        private final static String INFOXX = "orderlog.info.xx";
        private final static String WARNING = "productlog.warning";
    
        public static void main(String[] argv) throws Exception { 
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            // 从连接中创建通道
            Channel channel = connection.createChannel();
    
            //定义交换机
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
    
            //在收到消费者响应之前，最多发送多少条消息给消费者
            //用于能者多劳模式
            channel.basicQos(1);
    
            // 消息内容. 这里error，info，warning的消息各发送2个
            channel.basicPublish(EXCHANGE_NAME, ERROR, null, "Hello rabbitmq! error1".getBytes());
            channel.basicPublish(EXCHANGE_NAME, ERROR, null, "Hello rabbitmq! error2".getBytes());
            channel.basicPublish(EXCHANGE_NAME, INFO, null, "Hello rabbitmq! info1".getBytes());
            channel.basicPublish(EXCHANGE_NAME, INFO, null, "Hello rabbitmq! info2".getBytes());
            channel.basicPublish(EXCHANGE_NAME, INFOXX, null, "Hello rabbitmq! info1XX".getBytes());
            channel.basicPublish(EXCHANGE_NAME, INFOXX, null, "Hello rabbitmq! info2XX".getBytes());
            channel.basicPublish(EXCHANGE_NAME, WARNING, null, "Hello rabbitmq! warning1".getBytes());
            channel.basicPublish(EXCHANGE_NAME, WARNING, null, "Hello rabbitmq! warning2".getBytes());
    
            //关闭通道和连接
            channel.close();
            connection.close();
        }
    }
    
    public class Recv1 { 
        private final static String EXCHANGE_NAME = "q_test_topic_exchange";
        private final static String QUEUE_NAME = "q_test_topic1";
    
        public static void main(String[] argv) throws Exception { 
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            // 从连接中创建通道
            Channel channel = connection.createChannel();
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, true, false, false, null);
            //将队列绑定到交换机上
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "orderlog.*");
    
            channel.basicQos(1);
    
            //channel监听队列。
            channel.basicConsume(QUEUE_NAME, false, new DefaultConsumer(channel) { 
                //回调函数。队列中有数据后会回调该方法
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                                           byte[] body) throws IOException { 
                    try { 
                        String message = new String(body, "UTF-8");
                        System.out.println(" [1] Received : " + message);
                    } finally { 
                        //手动提交ack回执
                        channel.basicAck(envelope.getDeliveryTag(), false);
                    }
                }
            });
        }
    }
    
    public class Recv2 { 
        private final static String EXCHANGE_NAME = "q_test_topic_exchange";
        private final static String QUEUE_NAME = "q_test_topic2";
    
        public static void main(String[] argv) throws Exception { 
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            // 从连接中创建通道
            Channel channel = connection.createChannel();
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, true, false, false, null);
            //将队列绑定到交换机上
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "orderlog.error");
    
            channel.basicQos(1);
    
            //channel监听队列。
            channel.basicConsume(QUEUE_NAME, false, new DefaultConsumer(channel) { 
                //回调函数。队列中有数据后会回调该方法
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                                           byte[] body) throws IOException { 
                    try { 
                        String message = new String(body, "UTF-8");
                        System.out.println(" [2] Received : " + message);
                    } finally { 
                        //手动提交ack回执
                        channel.basicAck(envelope.getDeliveryTag(), false);
                    }
                }
            });
        }
    }
    
    public class Recv3 { 
        private final static String EXCHANGE_NAME = "q_test_topic_exchange";
        private final static String QUEUE_NAME = "q_test_topic3";
    
        public static void main(String[] argv) throws Exception { 
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            // 从连接中创建通道
            Channel channel = connection.createChannel();
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, true, false, false, null);
            //将队列绑定到交换机上.
            //*:只能向后或向前匹配一个词；#：可以匹配0个或多个
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "orderlog.#");
    
            channel.basicQos(1);
    
            //channel监听队列。
            channel.basicConsume(QUEUE_NAME, false, new DefaultConsumer(channel) { 
                //回调函数。队列中有数据后会回调该方法
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                                           byte[] body) throws IOException { 
                    try { 
                        String message = new String(body, "UTF-8");
                        System.out.println(" [3] Received : " + message);
                    } finally { 
                        //手动提交ack回执
                        channel.basicAck(envelope.getDeliveryTag(), false);
                    }
                }
            });
        }
    }

首先运行三个消费者，将队列绑定到交换器上，然后启动生产者发送消息。  
控制台输出如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 18]  
![在这里插入图片描述][20200221135627367.png]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 19]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200219224914942.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp,size_16,color_FFFFFF,t_70
[20200219224922976.png]: https://img-blog.csdnimg.cn/20200219224922976.png
[20200219204124553.png]: https://img-blog.csdnimg.cn/20200219204124553.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200219203828512.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/2020021920442883.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/2020021920460855.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp,size_16,color_FFFFFF,t_70
[20200219210141801.png]: https://img-blog.csdnimg.cn/20200219210141801.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/20200219210241757.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp,size_16,color_FFFFFF,t_70
[20200219210518237.png]: https://img-blog.csdnimg.cn/20200219210518237.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 5]: https://img-blog.csdnimg.cn/20200219212008125.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 6]: https://img-blog.csdnimg.cn/20200219212013426.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 7]: https://img-blog.csdnimg.cn/20200219214225267.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 8]: https://img-blog.csdnimg.cn/2020021921423284.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp,size_16,color_FFFFFF,t_70
[20200219221036411.png]: https://img-blog.csdnimg.cn/20200219221036411.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 9]: https://img-blog.csdnimg.cn/20200219224106496.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 10]: https://img-blog.csdnimg.cn/20200219224111965.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 11]: https://img-blog.csdnimg.cn/20200219224116766.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 12]: https://img-blog.csdnimg.cn/20200219224341957.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 13]: https://img-blog.csdnimg.cn/20200219224413528.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 14]: https://img-blog.csdnimg.cn/2020021922450723.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp,size_16,color_FFFFFF,t_70
[20200219225416754.png]: https://img-blog.csdnimg.cn/20200219225416754.png
[20200219230701966.png]: https://img-blog.csdnimg.cn/20200219230701966.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 15]: https://img-blog.csdnimg.cn/20200219230944590.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 16]: https://img-blog.csdnimg.cn/20200219230950160.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 17]: https://img-blog.csdnimg.cn/20200221134727903.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp,size_16,color_FFFFFF,t_70
[2020022113504116.png]: https://img-blog.csdnimg.cn/2020022113504116.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 18]: https://img-blog.csdnimg.cn/20200221135621572.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp,size_16,color_FFFFFF,t_70
[20200221135627367.png]: https://img-blog.csdnimg.cn/20200221135627367.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 19]: https://img-blog.csdnimg.cn/20200221135634499.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp,size_16,color_FFFFFF,t_70