RabbitMQ入门

RabbitMQ入门

目录

MQ介绍

MQ的优势

RabbitMQ是什么

AMQP

 RabbitMQ的相关概念

安装及配置RabbitMQ

RabbitMQ的六种工作模式

简单模式

示例工程搭建

工作模式

Publish/Subscribe发布与订阅模式

 Routing 路由模式

Topic 通配符模式

 总结

---

---

MQ介绍

MQ全称 Message Queue（消息队列），是在消息的传输过程中保存消息的容器。多用于分布式系统之间进行通信。

未加入MQ时，应用之间的远程调用

加入MQ后应用之间的调用

MQ的优势

1. 应用解耦：MQ相当于一个中介，生产方通过MQ与消费方交互，它将应用程序进行解耦合，从而提升容错性和可维护性。
2. 任务异步处理：将不需要同步处理的并且耗时长的 *** 作由消息队列通知消息接收方进行异步处理。提高了应用程序的响应时间。
3. 削峰填谷：如订单系统，在下单的时候就会往数据库写数据。但是数据库只能支撑每秒1000左右的并发写入，并发量再高就容易宕机。低峰期的时候并发也就100多个，但是在高峰期时候，并发量会突然激增到5000以上，这个时候数据库肯定卡死了。消息被MQ保存起来了，然后系统就可以按照自己的消费能力来消费，比如每秒1000个消息，这样慢慢
   写入数据库，这样就不会卡死数据库了。

RabbitMQ是什么

RabbitMQ是消息队列中的一种，是由 LShift 提供的一个 Advanced Message Queuing Protocol (AMQP) 的开源实现，以高性能、健壮以及可伸缩性出名的 Erlang 写成。

AMQP

AMQP，即 Advanced Message Queuing Protocol（高级消息队列协议），是一个网络协议，是应用层协议的一个开放标准，为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，遵循此协议，不收客户端和中间件产品和开发语言限制。2006年，AMQP 规范发布。类比HTTP。

 RabbitMQ的相关概念

• Broker：接收和分发消息的应用，RabbitMQ Server就是 Message Broker
• Virtual host：出于多租户和安全因素设计的，把 AMQP 的基本组件划分到一个虚拟的分组中，类似于网络中的 namespace 概念。当多个不同的用户使用同一个 RabbitMQ server 提供的服务时，可以划分出多个vhost，每个用户在自己的 vhost 创建 exchange／queue 等
• Connection：publisher／consumer 和 broker 之间的 TCP 连接
• Channel：如果每一次访问 RabbitMQ 都建立一个 Connection，在消息量大的时候建立 TCP Connection的开销将是巨大的，效率也较低。Channel 是在 connection 内部建立的逻辑连接，如果应用程序支持多线程，通常每个thread创建单独的 channel 进行通讯，AMQP method 包含了channel id 帮助客户端和message broker 识别 channel，所以 channel 之间是完全隔离的。Channel 作为轻量级的 Connection 极大减少了 *** 作系统建立 TCP connection 的开销
• Exchange：message 到达 broker 的第一站，根据分发规则，匹配查询表中的 routing key，分发消息到queue 中去。常用的类型有：direct (point-to-point), topic (publish-subscribe) 和fanout (multicast)
• Queue：消息最终被送到这里等待 consumer 取走
• Binding：exchange 和 queue 之间的虚拟连接，binding 中可以包含 routing key。Binding 信息被保存到 exchange 中的查询表中，用于 message 的分发依据

安装及配置RabbitMQ

一般来说安装 RabbitMQ 之前要安装 Erlang ，可以去Erlang官网下载。接着去RabbitMQ官网下载安装包，之后解压缩即可。根据 *** 作系统不同官网提供了相应的安装说明：Windows、Debian / Ubuntu、RPM-based Linux、Mac，也可参考网上教程进行安装。

RabbitMQ的六种工作模式

RabbitMQ提供了6种模式：简单模式，work模式，Publish/Subscribe发布与订阅模式，Routing路由模式，Topics主题模式，RPC远程调用模式（远程调用，不太算MQ；暂不作介绍）。
官网对应模式介绍：https://www.rabbitmq.com/getstarted.html

简单模式

• P：生产者，也就是要发送消息的程序
• C：消费者：消息的接收者，会一直等待消息到来
• queue：消息队列，图中红色部分。类似一个邮箱，可以缓存消息；生产者向其中投递消息，消费者从其中取出消息

示例工程搭建

1. 创建 Maven 项目
2. 导入依赖

   
       
           com.rabbitmq
           amqp-client
           5.6.0
       
   

3. 编写工具类

   public class RabbitMqUtil {
       private Connection connection;
       private Channel channel;
   
       public RabbitMqUtil() {
       }
   
       
       public Channel getChannel() throws IOException, TimeoutException {
           //创建连接
           ConnectionFactory connectionFactory = new ConnectionFactory();
           //设置主机地址，默认为 localhost
           connectionFactory.setHost("192.168.137.5");
           //设置连接端口，默认为 5672
           connectionFactory.setPort(5672);
           //设置虚拟主机名称，默认为 /
           connectionFactory.setVirtualHost("/fgba");
           //设置连接用户名，默认为 guest
           connectionFactory.setUsername("fgba");
           //设置连接密码，默认为 guest
           connectionFactory.setPassword("123456");
           //创建频道
           connection = connectionFactory.newConnection();
           //声明（创建）队列
           channel = connection.createChannel();
           return channel;
       }
   
       
       public void close() throws IOException, TimeoutException {
           channel.close();
           connection.close();
       }
   }
   

4. 创建 simple 包
5. 在包中编写生产者

   public class Producer {
       public static String queueName = "simple_queue";
   
       public static void main(String[] args) throws IOException, TimeoutException {
           RabbitMqUtil rabbitMqUtil = new RabbitMqUtil();
           Channel channel = rabbitMqUtil.getChannel();
           
           channel.queueDeclare(queueName,true,false,false,null);
           //发送消息
           String message = "你好，RabbitMQ";
           channel.basicPublish("", queueName,null,message.getBytes());
           System.out.println("消息已发送：" + message);
           //释放资源
           rabbitMqUtil.close();
       }
   }

6. 在包中编写消费者

   public class Consumer {
       public static void main(String[] args) throws IOException, TimeoutException {
           RabbitMqUtil rabbitMqUtil = new RabbitMqUtil();
           Channel channel = rabbitMqUtil.getChannel();
           
           channel.queueDeclare(Producer.queueName, true, false, false, null);
           //创建消费者；并设置消息处理
           Consumer consumer = new DefaultConsumer(channel) {
               
               @Override
               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("接收到的消息为：" + new String(body, StandardCharsets.UTF_8));
               }
           };
   
           
           channel.basicConsume(Producer.queueName,true, consumer);
           //不关闭资源，应该一直监听消息
       }
   }

7. 运行生产者

    进入RabbitMQ控制台查看

8. 运行消费者接收消息

工作模式

Work Queues 与入门程序的简单模式相比，多了一个或一些消费端，多个消费端共同消费同一个队列中的消息。
应用场景：对于任务过重或任务较多情况使用工作队列可以提高任务处理的速度。

代码示例

1. 在上文的工程中新建work包
2. 编写生产者

   public class Producer {
       public static String queueName = "work_queue";
   
       public static void main(String[] args) throws IOException, TimeoutException {
           RabbitMqUtil rabbitMqUtil = new RabbitMqUtil();
           Channel channel = rabbitMqUtil.getChannel();
           channel.queueDeclare(queueName, true, false, false, null);
           for (int i = 0; i < 30; i++) {
               //发送消息
               String message = "你好，RabbitMQ - work-queue - " + (i + 1);
               channel.basicPublish("", queueName, null, message.getBytes());
               System.out.println("消息已发送：" + message);
           }
   
           //释放资源
           rabbitMqUtil.close();
       }
   }

3. 编写消费者1

   public class Consumer1 {
       public static void main(String[] args) throws IOException, TimeoutException {
           RabbitMqUtil rabbitMqUtil = new RabbitMqUtil();
           Channel channel = rabbitMqUtil.getChannel();
           // 声明（创建）队列
           
           channel.queueDeclare(Producer.queueName, 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 {
                   //路由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, StandardCharsets.UTF_8));
               }
           };
   
           
           channel.basicConsume(Producer.queueName,true, consumer);
           //不关闭资源，应该一直监听消息
       }
   }

4. 编写消费者2

   public class Consumer2 {
       public static void main(String[] args) throws IOException, TimeoutException {
           RabbitMqUtil rabbitMqUtil = new RabbitMqUtil();
           Channel channel = rabbitMqUtil.getChannel();
           // 声明（创建）队列
           
           channel.queueDeclare(Producer.queueName, 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 {
                   //路由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, StandardCharsets.UTF_8));
               }
           };
   
           
           channel.basicConsume(Producer.queueName,true, consumer);
           //不关闭资源，应该一直监听消息
       }
   }
   

5. 运行消费者1和消费者2监听消息

6. 运行生产者

7. 查看消费者1和消费者2

   可以看出，两个消费者竞争性地接收消息，即如果队列存在多个消费者，则消费者对于同一个消息的关系是竞争的关系

Publish/Subscribe发布与订阅模式

发布订阅模式：

1、每个消费者监听自己的队列。

2、生产者将消息发给broker，由交换机将消息转发到绑定此交换机的每个队列，每个绑定交换机的队列都将接收到消息

1. 在上文的工程中新建 ps 包
2. 编写生产者

   public class Producer {
       public static String fanoutExchange = "fanout_exchange";
       public static String fanoutQueue1 = "fanout_queue_1";
       public static String fanoutQueue2 = "fanout_queue_2";
   
   
       public static void main(String[] args) throws IOException, TimeoutException {
           RabbitMqUtil rabbitMqUtil = new RabbitMqUtil();
           Channel channel = rabbitMqUtil.getChannel();
   
           for (int i = 0; i < 10; i++) {
               //发送消息
               String message = "你好，RabbitMQ - fanout-queue - " + (i + 1);
               channel.basicPublish(fanoutExchange, "", null, message.getBytes());
               System.out.println("消息已发送：" + message);
           }
   
           //释放资源
           rabbitMqUtil.close();
       }
   }

3. 编写消费者1

   public class Consumer1 {
       public static void main(String[] args) throws IOException, TimeoutException {
           RabbitMqUtil rabbitMqUtil = new RabbitMqUtil();
           Channel channel = rabbitMqUtil.getChannel();
           
           channel.exchangeDeclare(Producer.fanoutExchange, BuiltinExchangeType.FANOUT);
           // 声明（创建）队列
           
           channel.queueDeclare(Producer.fanoutQueue1, true, false, false, null);
           //队列绑定交换机
           channel.queueBind(Producer.fanoutQueue1, Producer.fanoutExchange, "");
   
           //创建消费者；并设置消息处理
           Consumer consumer = new DefaultConsumer(channel) {
               
               @Override
               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, StandardCharsets.UTF_8));
               }
           };
   
           
           channel.basicConsume(Producer.fanoutQueue1, true, consumer);
           //不关闭资源，应该一直监听消息
       }
   }

4. 编写消费者2

   public class Consumer2 {
       public static void main(String[] args) throws IOException, TimeoutException {
           RabbitMqUtil rabbitMqUtil = new RabbitMqUtil();
           Channel channel = rabbitMqUtil.getChannel();
            
           channel.exchangeDeclare(Producer.fanoutExchange, BuiltinExchangeType.FANOUT);
           // 声明（创建）队列
           
           channel.queueDeclare(Producer.fanoutQueue2, true, false, false, null);
           //队列绑定交换机
           channel.queueBind(Producer.fanoutQueue2, Producer.fanoutExchange, "");
   
           //创建消费者；并设置消息处理
           Consumer consumer = new DefaultConsumer(channel) {
               
               @Override
               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, StandardCharsets.UTF_8));
               }
           };
   
           
           channel.basicConsume(Producer.fanoutQueue2, true, consumer);
           //不关闭资源，应该一直监听消息
       }
   }

5. 启动消费者1和消费者2
6. 运行生产者
7. 查看消费者1和消费者2

         由这里可以看到两边都接收到了生产者发来的消息，即达到了广播的效果

总结：发布订阅模式主要是先通过交换机绑定队列，然后生产者发送给交换机，交换机在收到消息时会把消息转发给绑定的队列

发布订阅模式与工作队列模式的区别
1、工作队列模式不用定义交换机，而发布/订阅模式需要定义交换机。
2、发布/订阅模式的生产方是面向交换机发送消息，工作队列模式的生产方是面向队列发送消息(底层使用默认交换机)。
3、发布/订阅模式需要设置队列和交换机的绑定，工作队列模式不需要设置，实际上工作队列模式会将队列绑定到默认的交换机 

 Routing 路由模式

 路由模式特点：

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

代码示例

1. 在上文的工程中新建 routing 包
2. 编写生产者

   public class Producer {
       
       public static String directExchange = "direct_exchange";
       public static String directQueueInsert = "direct_queue_insert";
       public static String directQueueUpdate = "direct_queue_update";
   
   
       public static void main(String[] args) throws IOException, TimeoutException {
           RabbitMqUtil rabbitMqUtil = new RabbitMqUtil();
           Channel channel = rabbitMqUtil.getChannel();
   
           //发送新增消息
           String insertMessage = "routing-queue - routing key:insert - 新增商品";
           channel.basicPublish(directExchange, "insert", null, insertMessage.getBytes());
           System.out.println("消息已发送：" + insertMessage);
   
           //发送修改消息
           String updateMessage = "routing-queue - routing key:update - 修改商品";
           channel.basicPublish(directExchange, "update", null, updateMessage.getBytes());
           System.out.println("消息已发送：" + updateMessage);
           //释放资源
           rabbitMqUtil.close();
       }
   }

3. 编写消费者1

   public class Consumer1 {
       public static void main(String[] args) throws IOException, TimeoutException {
           RabbitMqUtil rabbitMqUtil = new RabbitMqUtil();
           Channel channel = rabbitMqUtil.getChannel();
           
           channel.exchangeDeclare(Producer.directExchange, BuiltinExchangeType.DIRECT);
           // 声明（创建）队列
           
           channel.queueDeclare(Producer.directQueueInsert, true, false, false, null);
           //队列绑定交换机
           channel.queueBind(Producer.directQueueInsert, Producer.directExchange,"insert");
   
           //创建消费者；并设置消息处理
           Consumer consumer = new DefaultConsumer(channel) {
               
               @Override
               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, StandardCharsets.UTF_8));
               }
           };
   
           
           channel.basicConsume(Producer.directQueueInsert,true, consumer);
           //不关闭资源，应该一直监听消息
       }
   }

4. 编写消费者2

   public class Consumer2 {
       public static void main(String[] args) throws IOException, TimeoutException {
           RabbitMqUtil rabbitMqUtil = new RabbitMqUtil();
           Channel channel = rabbitMqUtil.getChannel();
           
           channel.exchangeDeclare(Producer.directExchange, BuiltinExchangeType.DIRECT);
           // 声明（创建）队列
           
           channel.queueDeclare(Producer.directQueueUpdate, true, false, false, null);
           //队列绑定交换机
           channel.queueBind(Producer.directQueueUpdate, Producer.directExchange,"update");
   
           //创建消费者；并设置消息处理
           com.rabbitmq.client.Consumer consumer = new DefaultConsumer(channel) {
               
               @Override
               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, StandardCharsets.UTF_8));
               }
           };
   
           
           channel.basicConsume(Producer.directQueueUpdate,true, consumer);
           //不关闭资源，应该一直监听消息
       }
   }

5. 启动消费者1和消费者2
6. 运行生产者
7. 查看消费者1和消费者2

         可以看出两个消费者的路由 key 不同，接收的消息也不相同

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

Topic 通配符模式

Topic 类型与 Direct 相比，都是可以根据 RoutingKey 把消息路由到不同的队列。只不过 Topic 类型 Exchange 可以让队列在绑定 Routing key 的时候使用通配符！
Routingkey 一般都是有一个或多个单词组成，多个单词之间以“.”分割，例如： item.insert

通配符规则：

• #：匹配一个或多个词
• *：匹配不多不少恰好1个词

举例：

• item.#：能够匹配 item.insert.abc 或者 item.insert
• item.*：只能匹配 item.insert

 

红色Queue：绑定的是 usa.# ，因此凡是以 usa.开头的 routing key 都会被匹配到

黄色Queue：绑定的是 #.news ，因此凡是以 .news结尾的 routing key 都会被匹配 

代码示例

1. 在上文的工程中新建 topic 包
2. 编写生产者

   public class Producer {
       public static String topicExchange = "topic_exchange";
       public static String topicQueueAll = "topic_queue_all";
       public static String topicQueueUpdateInsert = "topic_queue_update_insert";
   
   
       public static void main(String[] args) throws IOException, TimeoutException {
           RabbitMqUtil rabbitMqUtil = new RabbitMqUtil();
           Channel channel = rabbitMqUtil.getChannel();
           
           channel.exchangeDeclare(topicExchange, BuiltinExchangeType.TOPIC);
   
           //发送新增消息
           String insertMessage = "topic-queue - routing key:item.insert - 新增商品";
           channel.basicPublish(topicExchange, "item.insert", null, insertMessage.getBytes());
           System.out.println("消息已发送：" + insertMessage);
   
           //发送修改消息
           String updateMessage = "topic-queue - routing key:item.update - 修改商品";
           channel.basicPublish(topicExchange, "item.update", null, updateMessage.getBytes());
           System.out.println("消息已发送：" + updateMessage);
   
           //发送修改消息
           String deleteMessage = "topic-queue - routing key:item.delete - 删除商品";
           channel.basicPublish(topicExchange, "item.delete", null, deleteMessage.getBytes());
           System.out.println("消息已发送：" + deleteMessage);
   
           //释放资源
           rabbitMqUtil.close();
       }
   }

3. 编写消费者1

   public class Consumer1 {
       public static void main(String[] args) throws IOException, TimeoutException {
           RabbitMqUtil rabbitMqUtil = new RabbitMqUtil();
           Channel channel = rabbitMqUtil.getChannel();
           
           channel.exchangeDeclare(Producer.topicExchange, BuiltinExchangeType.TOPIC);
           // 声明（创建）队列
           
           channel.queueDeclare(Producer.topicQueueAll, true, false, false, null);
           //队列绑定交换机
           channel.queueBind(Producer.topicQueueAll, Producer.topicExchange, "item.*");
   
           //创建消费者；并设置消息处理
           com.rabbitmq.client.Consumer consumer = new DefaultConsumer(channel) {
               
               @Override
               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, StandardCharsets.UTF_8));
               }
           };
   
           
           channel.basicConsume(Producer.topicQueueAll, true, consumer);
           //不关闭资源，应该一直监听消息
       }
   }

4. 编写消费者2

   public class Consumer2 {
       public static void main(String[] args) throws IOException, TimeoutException {
           RabbitMqUtil rabbitMqUtil = new RabbitMqUtil();
           Channel channel = rabbitMqUtil.getChannel();
           
           channel.exchangeDeclare(Producer.topicExchange, BuiltinExchangeType.TOPIC);
           // 声明（创建）队列
           
           channel.queueDeclare(Producer.topicQueueUpdateInsert, true, false, false, null);
           //队列绑定交换机
           channel.queueBind(Producer.topicQueueUpdateInsert, Producer.topicExchange,"item.update");
           channel.queueBind(Producer.topicQueueUpdateInsert, Producer.topicExchange,"item.insert");
   
           //创建消费者；并设置消息处理
           com.rabbitmq.client.Consumer consumer = new DefaultConsumer(channel) {
               
               @Override
               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, StandardCharsets.UTF_8));
               }
           };
   
           
           channel.basicConsume(Producer.topicQueueUpdateInsert,true, consumer);
           //不关闭资源，应该一直监听消息
       }
   }

5. 启动消费者1和消费者2
6. 运行生产者
7. 查看消费者1和消费者2

        可以看到消费者1接收到了增加商品、修改商品和删除商品的消息，而消费者1接收到了增加商品、修改商品的消息，我们再来看消费者1的routing key

其 routing key “item.*” 匹配 item.1个单词，我们发送的消息都匹配这一规则，所以消费者1接收到了所有信息，而消费者2没有使用通配符，而是分别绑定了2个routing key “item.update”和“item.insert”，所以只收到了更新消息和插入消息

 总结：Topic主题模式可以实现 Publish/Subscribe 发布与订阅模式和 Routing 路由模式的功能；只是 Topic 在配置 routing key 的时候可以使用通配符，显得更加灵活。

总结

本文介绍了MQ和MQ的优势，以及RabbitMQ的基本概念和以下5种RabbitMQ 工作模式：

1、简单模式 HelloWorld：一个生产者、一个消费者，不需要设置交换机（使用默认的交换机）
2、工作队列模式 Work Queue：一个生产者、多个消费者（竞争关系），不需要设置交换机（使用默认的交换机）
3、发布订阅模式 Publish/subscribe：需要设置类型为 fanout 的交换机，并且交换机和队列进行绑定，当发送消息到交换机后，交换机会将消息发送到绑定的队列
4、路由模式 Routing ：需要设置类型为 direct 的交换机，交换机和队列进行绑定，并且指定 routing key，当发送消息到交换机后，交换机会根据 routing key 将消息发送到对应的队列
5、通配符模式 Topic ：需要设置类型为 topic 的交换机，交换机和队列进行绑定，并且指定通配符方式的 routing key，当发送消息到交换机后，交换机会根据 routing key 将消息发送到对应的队列