RabbitMQ基础-工作模式

提示：RabbitMQ五种工作模式及应用（官方是7种）
  
  

---

媒介

RabbitMQ官方文档

---

一、RabbitMQ安装（Docker）

• 下载镜像：
  
  1. docker pull rabbitmq

  复制代码
• 运行容器：
  
  1. docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=123456 rabbitmq:latest

  复制代码
  账号admin，暗码123456 用于登录MQ后台
  
• 打开访问界面
  
  1. 进入容器：
  2. docker exec -it a64bbac6d26f bash
  3. 执行：
  4. rabbitmq-plugins enable rabbitmq_management
  5. 输出：
  6. Enabling plugins on node rabbit@a64bbac6d26f:
  7. rabbitmq_management
  8. The following plugins have been configured:
  9. rabbitmq_management
  10. rabbitmq_management_agent
  11. rabbitmq_prometheus
  12. rabbitmq_web_dispatch
  13. Applying plugin configuration to rabbit@a64bbac6d26f...
  14. The following plugins have been enabled:
  15. rabbitmq_management
  16. started 1 plugins

  复制代码
• 即可在浏览器中通过 http://127.0.0.1:15672/ 访问，可通过上面设置的账号暗码进行登录。
  

二、MQ中相关概念

KEYVALUEBroker接收和分发消息的应用Virtual host出于多租户和安全因素计划的，把 AMQP 的根本组件分别到一个虚拟的分组中，雷同于网络中的 namespace 概念。当多个差别的用户使用同一个 RabbitMQ server 提供的服务时，可以分别出多个vhost，每个用户在自己的 vhost 创建 exchange／queue 等Connectionpublisher／consumer 和 broker 之间的 TCP 连接Channel假如每一次访问 RabbitMQ 都建立一个 Connection，在消息量大的时候建立 TCP Connection的开销将是巨大的，服从也较低。Channel 是在 connection 内部建立的逻辑连接，假如应用步伐支持多线程，通常每个thread创建单独的 channel 进行通讯，AMQP method 包罗了channel id 帮助客户端和message broker 识别 channel，以是 channel 之间是完全隔离的。Channel 作为轻量级的 Connection 极大减少了操作体系建立 TCP connection 的开销Exchangemessage 到达 broker 的第一站，根据分发规则，匹配查询表中的 routing key，分发消息到queue 中去。常用的范例有：direct (定向，把消息交给符合指定routing key 的队列), topic (通配符，把消息交给符合routing pattern（路由模式） 的队列) and fanout (广播，将消息交给所有绑定到交换机的队列)Queue消息最终被送到这里等待 consumer 取走Bindingexchange 和 queue 之间的虚拟连接，binding 中可以包罗 routing key。Binding 信息被生存到 exchange 中的查询表中，用于 message 的分发依据 每个Virtual Hosts相当于一个相对独立的RabbitMQ服务器，每个VirtualHost之间是相互隔离的。exchange、queue、message不能互通。 相当于mysql的db。Virtual Name一般以/开头
Exchange（交换机）只负责转发消息，不具备存储消息的本领，因此假如没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失！
三、工作模式

模式特点场景图解简单 (Simple)一个生产者、一个消耗者，不必要设置交换机（使用默认的交换机）单个生产者和单个消耗者（日志记录，订单处理）

工作(Work Queue)一个生产者、多个消耗者（竞争关系），不必要设置交换机（使用默认的交换机）必要多个消耗者共同处理任务（异步处理，短信关照）

发布/订阅(Publish/Subscribe)必要设置范例为fanout的交换机，而且交换机和队列进行绑定，当发送消息到交换机后，交换机会将消息发送到绑定的队列必要消息被多个消耗者同时接收（实时关照，广播）

路由（Routing）必要设置范例为direct的交换机，交换机和队列进行绑定，而且指定routingkey，当发送消息到交换机后，交换机会根据routing key将消息发送到对应的队列必要根据特定条件匹配路由消息（消息分类处理）

通配符(Topics)必要设置范例为topic的交换机，交换机和队列进行绑定，而且指定通配符方式的routing key，当发送消息到交换机后，交换机会根据routing key将消息发送到对应的队列适用于必要根据模式匹配分发消息的场景

四、Java操作MQ

依赖：

1. <dependency>
2.      <groupId>com.rabbitmq</groupId>
3.      <artifactId>amqp-client</artifactId>
4.      <version>5.21.0</version>
5. </dependency>

复制代码

工具类：

1. package com.rabbitmq.utils;
2. import com.rabbitmq.client.ConnectionFactory;
3. import com.rabbitmq.client.Connection;
4. import java.io.IOException;
5. import java.util.concurrent.TimeoutException;
6. public class ConnectionUntils {
7.     public static Connection getConnection() throws IOException, TimeoutException {
8.         //创建连接工厂
9.         ConnectionFactory factory = new ConnectionFactory();
10.         factory.setHost("127.0.0.1");
11.         factory.setPort(5672);
12.         factory.setVirtualHost("/");
13.         factory.setUsername("admin");
14.         factory.setPassword("123456");
15.         //创建连接
16.         Connection connection = factory.newConnection();
17.         return connection;
18.     }
19. }

复制代码

• 简单模式
  

1. package com.rabbitmq.simple;
2. import com.rabbitmq.client.Channel;
3. import com.rabbitmq.client.Connection;
4. import com.rabbitmq.client.ConnectionFactory;
5. import com.rabbitmq.utils.ConnectionUntils;
6. /**
7. * 简单模式
8. * 生产者
9. */
10. public class Producer {
11.     static final String QUEUE_NAME = "simple_name";
12.     public static void main(String[] args) throws Exception {
13.         Connection connection = ConnectionUntils.getConnection();
14.         Channel channel = connection.createChannel();
15.         /**
16.          * 声明队列
17.          * 参数1：队列名称
18.          * 参数2：是否定义持久化队列
19.          * 参数3：是否独占本次连接
20.          * 参数4：是否在不使用的时候自动删除队列
21.          * 参数5：队列其它参数
22.          */
23.         channel.queueDeclare(QUEUE_NAME, true, false, false, null);
24.         /**
25.          * 发送消息
26.          * 参数1：交换机名称，如果没有指定则使用默认Default Exchage
27.          * 参数2：路由key,简单模式可以传递队列名称
28.          * 参数3：消息其它属性
29.          * 参数4：消息内容
30.          */
31.         String message = "Hello World!7";
32.         channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
33.         System.out.println(" [x] Sent '" + message + "'");
34.         //释放资源
35.         channel.close();
36.         connection.close();
37.     }
38. }

复制代码

1. package com.rabbitmq.simple;
2. import com.rabbitmq.client.*;
3. import com.rabbitmq.utils.ConnectionUntils;
4. import java.io.IOException;
5. /**
6. * 消费者
7. */
8. public class Consumer {
9.     public static void main(String[] args) throws Exception {
10.         Connection conection = ConnectionUntils.getConnection();
11.         //通道
12.         Channel channel = conection.createChannel();
13.         /**
14.          * 声明队列
15.          * 参数1：队列名称
16.          * 参数2：是否定义持久化队列
17.          * 参数3：是否独占本次连接
18.          * 参数4：是否在不使用的时候自动删除队列
19.          * 参数5：队列其它参数
20.          */
21.         channel.queueDeclare(Producer.QUEUE_NAME, true, false, false, null);
22.         //接受消息
23.         DefaultConsumer consumer = new DefaultConsumer(channel) {
24.             /**
25.              * 接受消息 处理消息的回调函数
26.              * @param consumerTag 消费者标签 channel.basicConsume() 指定使用什么标签的消费者消费消息
27.              * @param envelope 消息包的信息：消息id routingkey exchange
28.              * @param properties 属性信息
29.              * @param body 消息
30.              * @throws IOException
31.              */
32.             @Override
33.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
34.                 System.out.println("路由：" + envelope.getRoutingKey());
35.                 System.out.println("交换机：" + envelope.getExchange());
36.                 System.out.println("消息ID：" + envelope.getDeliveryTag());
37.                 System.out.println("消息" + new String(body, "UTF-8"));
38.                 super.handleDelivery(consumerTag, envelope, properties, body);
39.             }
40.         };
41.         /**
42.          * 1.队列名称
43.          * 2.是否确认，true 消息消费后，mq自动回复消息已消费 mq收到回复后删除消息
44.          * 3.消费者对象，消息消费的回调函数
45.          */
46.         channel.basicConsume(Producer.QUEUE_NAME, true, consumer);
47.     }
48. }

复制代码

• 工作模式:在一个队列中假如有多个消耗者，那么消耗者之间对于同一个消息的关系是竞争的关系。
  

1. package com.rabbitmq.work;
2. import com.rabbitmq.client.Channel;
3. import com.rabbitmq.client.Connection;
4. import com.rabbitmq.utils.ConnectionUntils;
5. /**
6. * 工作模式
7. * 生产者
8. */
9. public class Producer {
10.     static final String QUEUE_NAME = "work_name";
11.     public static void main(String[] args) throws Exception {
12.         Connection connection = ConnectionUntils.getConnection();
13.         Channel channel = connection.createChannel();
14.         channel.queueDeclare(QUEUE_NAME, true, false, false, null);
15.         for (int i = 1; i < 30; i++) {
16.             String message = "Hello World!" + i;
17.             channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
18.             System.out.println(" [x] Sent '" + message + "'");
19.         }
20.         channel.close();
21.         connection.close();
22.     }
23. }

复制代码

1. package com.rabbitmq.work;
2. import com.rabbitmq.client.*;
3. import com.rabbitmq.utils.ConnectionUntils;
4. import java.io.IOException;
5. /**
6. * 消费者1
7. */
8. public class Consumer1 {
9.     public static void main(String[] args) throws Exception {
10.         Connection conection = ConnectionUntils.getConnection();
11.         Channel channel = conection.createChannel();
12.         channel.queueDeclare(Producer.QUEUE_NAME, true, false, false, null);
13.         DefaultConsumer consumer = new DefaultConsumer(channel) {
14.             @Override
15.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
16.                 System.out.println("路由：" + envelope.getRoutingKey());
17.                 System.out.println("交换机：" + envelope.getExchange());
18.                 System.out.println("消息ID：" + envelope.getDeliveryTag());
19.                 System.out.println("消息" + new String(body, "UTF-8"));
20.                 super.handleDelivery(consumerTag, envelope, properties, body);
21.             }
22.         };
23.         channel.basicConsume(Producer.QUEUE_NAME, true, consumer);
24.     }
25. }

复制代码

1. package com.rabbitmq.work;
2. import com.rabbitmq.client.*;
3. import com.rabbitmq.utils.ConnectionUntils;
4. import java.io.IOException;
5. /**
6. * 消费者2
7. */
8. public class Consumer2 {
9.     public static void main(String[] args) throws Exception {
10.         Connection conection = ConnectionUntils.getConnection();
11.         Channel channel = conection.createChannel();
12.         channel.queueDeclare(Producer.QUEUE_NAME, true, false, false, null);
13.         //接受消息
14.         DefaultConsumer consumer = new DefaultConsumer(channel) {
15.             @Override
16.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
17.                 System.out.println("路由：" + envelope.getRoutingKey());
18.                 System.out.println("交换机：" + envelope.getExchange());
19.                 System.out.println("消息ID：" + envelope.getDeliveryTag());
20.                 System.out.println("消息" + new String(body, "UTF-8"));
21.                 super.handleDelivery(consumerTag, envelope, properties, body);
22.             }
23.         };
24.         channel.basicConsume(Producer.QUEUE_NAME, true, consumer);
25.     }
26. }

复制代码

• 发布/订阅
  

1. package com.rabbitmq.ps;
3. import com.rabbitmq.client.BuiltinExchangeType;
4. import com.rabbitmq.client.Channel;
5. import com.rabbitmq.client.Connection;
6. import com.rabbitmq.utils.ConnectionUntils;
8. /**
9. * 发布订阅者模式
10. * 生产者
11. */
12. public class Producer {
13.     static final String EXCHANGE_NAME = "fanout_exchange";
14.     static final String QUEUE_NAME1 = "fanout_name_1";
15.     static final String QUEUE_NAME2 = "fanout_name_2";
16.     public static void main(String[] args) throws Exception {
17.         Connection connection = ConnectionUntils.getConnection();
18.         Channel channel = connection.createChannel();
19.         /**
20.          * 声明交换机
21.          * 1 交换机名称
22.          * 2 交换机类型 fanout
23.          *             direct
24.          *             topic
25.          */
26.         channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
27.         channel.queueDeclare(QUEUE_NAME1, true, false, false, null);
28.         channel.queueDeclare(QUEUE_NAME2, true, false, false, null);
29.         //队列绑定交换机
30.         channel.queueBind(QUEUE_NAME1,EXCHANGE_NAME,"");
31.         channel.queueBind(QUEUE_NAME2,EXCHANGE_NAME,"");
32.         for (int i = 1; i <= 10; i++) {
33.             String message = "Hello World!" + i;
34.             channel.basicPublish(EXCHANGE_NAME,"",null,message.getBytes());
35.             System.out.println(" [x] Sent '" + message + "'");
36.         }
37.         channel.close();
38.         connection.close();
39.     }
40. }

复制代码

1. package com.rabbitmq.ps;
2. import com.rabbitmq.client.*;
3. import com.rabbitmq.utils.ConnectionUntils;
4. import java.io.IOException;
5. /**
6. * 消费者1
7. */
8. public class Consumer1 {
9.     public static void main(String[] args) throws Exception {
10.         Connection conection = ConnectionUntils.getConnection();
11.         Channel channel = conection.createChannel();
12.         channel.queueDeclare(Producer.QUEUE_NAME1, true, false, false, null);
13.         channel.queueDeclare(Producer.QUEUE_NAME2, true, false, false, null);
14.         channel.queueBind(Producer.QUEUE_NAME1,Producer.EXCHANGE_NAME,"");
15.         channel.queueBind(Producer.QUEUE_NAME2,Producer.EXCHANGE_NAME,"");
16.         DefaultConsumer consumer = new DefaultConsumer(channel) {
17.             @Override
18.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
19.                 System.out.println("路由：" + envelope.getRoutingKey());
20.                 System.out.println("交换机：" + envelope.getExchange());
21.                 System.out.println("消息ID：" + envelope.getDeliveryTag());
22.                 System.out.println("消息" + new String(body, "UTF-8"));
23.                 super.handleDelivery(consumerTag, envelope, properties, body);
24.             }
25.         };
26.         channel.basicConsume(Producer.QUEUE_NAME1, true, consumer);
27.     }
28. }

复制代码

1. package com.rabbitmq.ps;
2. import com.rabbitmq.client.*;
3. import com.rabbitmq.utils.ConnectionUntils;
4. import java.io.IOException;
5. /**
6. * 消费者2
7. */
8. public class Consumer2 {
9.     public static void main(String[] args) throws Exception {
10.         Connection conection = ConnectionUntils.getConnection();
11.         Channel channel = conection.createChannel();
12.         channel.exchangeDeclare(Producer.EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
13.         channel.queueDeclare(Producer.QUEUE_NAME1, true, false, false, null);
14.         channel.queueDeclare(Producer.QUEUE_NAME2, true, false, false, null);
15.         channel.queueBind(Producer.QUEUE_NAME1,Producer.EXCHANGE_NAME,"");
16.         channel.queueBind(Producer.QUEUE_NAME2,Producer.EXCHANGE_NAME,"");
17.         DefaultConsumer consumer = new DefaultConsumer(channel) {
18.             @Override
19.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
20.                 System.out.println("路由：" + envelope.getRoutingKey());
21.                 System.out.println("交换机：" + envelope.getExchange());
22.                 System.out.println("消息ID：" + envelope.getDeliveryTag());
23.                 System.out.println("消息" + new String(body, "UTF-8"));
24.                 super.handleDelivery(consumerTag, envelope, properties, body);
25.             }
26.         };
27.         channel.basicConsume(Producer.QUEUE_NAME2, true, consumer);
28.     }
29. }

复制代码

• 路由模式
  

1. package com.rabbitmq.routing;
3. import com.rabbitmq.client.BuiltinExchangeType;
4. import com.rabbitmq.client.Channel;
5. import com.rabbitmq.client.Connection;
6. import com.rabbitmq.utils.ConnectionUntils;
8. /**
9. * 路由模式
10. * 生产者
11. */
12. public class Producer {
13.     static final String EXCHANGE_NAME = "direct_exchange";
14.     static final String QUEUE_NAME1 = "direct_name_insert";
15.     static final String QUEUE_NAME2 = "direct_name_update";
16.     public static void main(String[] args) throws Exception {
17.         Connection connection = ConnectionUntils.getConnection();
18.         Channel channel = connection.createChannel();
19.         channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
20.         channel.queueDeclare(QUEUE_NAME1, true, false, false, null);
21.         channel.queueDeclare(QUEUE_NAME2, true, false, false, null);
22.         channel.queueBind(QUEUE_NAME1,EXCHANGE_NAME,"insert");
23.         channel.queueBind(QUEUE_NAME2,EXCHANGE_NAME,"update");
24.         String message = "Hello World! - 新增 insert";
25.         channel.basicPublish(EXCHANGE_NAME,"insert",null,message.getBytes());
26.         System.out.println(" [x] Sent '" + message + "'");
27.         String message2 = "Hello World! - 修改 update";
28.         channel.basicPublish(EXCHANGE_NAME,"update",null,message2.getBytes());
29.         System.out.println(" [x] Sent '" + message2 + "'");
30.         channel.close();
31.         connection.close();
32.     }
33. }

复制代码

1. package com.rabbitmq.routing;
2. import com.rabbitmq.client.*;
3. import com.rabbitmq.utils.ConnectionUntils;
4. import java.io.IOException;
5. public class Consumer1 {
6.     public static void main(String[] args) throws Exception {
7.         Connection conection = ConnectionUntils.getConnection();
8.         Channel channel = conection.createChannel();
9.         channel.queueDeclare(Producer.QUEUE_NAME1, true, false, false, null);
10.         channel.queueDeclare(Producer.QUEUE_NAME2, true, false, false, null);
11.         channel.queueBind(Producer.QUEUE_NAME1,Producer.EXCHANGE_NAME,"insert");
12.         channel.queueBind(Producer.QUEUE_NAME2,Producer.EXCHANGE_NAME,"update");
13.         DefaultConsumer consumer = new DefaultConsumer(channel) {
14.             @Override
15.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
16.                 System.out.println("路由：" + envelope.getRoutingKey());
17.                 System.out.println("交换机：" + envelope.getExchange());
18.                 System.out.println("消息ID：" + envelope.getDeliveryTag());
19.                 System.out.println("消息" + new String(body, "UTF-8"));
20.                 super.handleDelivery(consumerTag, envelope, properties, body);
21.             }
22.         };
23.         channel.basicConsume(Producer.QUEUE_NAME1, true, consumer);
24.     }
25. }

复制代码

1. package com.rabbitmq.routing;
2. import com.rabbitmq.client.*;
3. import com.rabbitmq.utils.ConnectionUntils;
4. import java.io.IOException;
5. public class Consumer2 {
6.     public static void main(String[] args) throws Exception {
7.         Connection conection = ConnectionUntils.getConnection();
8.         Channel channel = conection.createChannel();
9.         channel.queueDeclare(Producer.QUEUE_NAME1, true, false, false, null);
10.         channel.queueDeclare(Producer.QUEUE_NAME2, true, false, false, null);
11.         channel.queueBind(Producer.QUEUE_NAME1,Producer.EXCHANGE_NAME,"insert");
12.         channel.queueBind(Producer.QUEUE_NAME2,Producer.EXCHANGE_NAME,"update");
13.         DefaultConsumer consumer = new DefaultConsumer(channel) {
14.             @Override
15.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
16.                 System.out.println("路由：" + envelope.getRoutingKey());
17.                 System.out.println("交换机：" + envelope.getExchange());
18.                 System.out.println("消息ID：" + envelope.getDeliveryTag());
19.                 System.out.println("消息" + new String(body, "UTF-8"));
20.                 super.handleDelivery(consumerTag, envelope, properties, body);
21.             }
22.         };
23.         channel.basicConsume(Producer.QUEUE_NAME2, true, consumer);
24.     }
25. }

复制代码

• 通配符模式
  

1. package com.rabbitmq.topic;
2. import com.rabbitmq.client.BuiltinExchangeType;
3. import com.rabbitmq.client.Channel;
4. import com.rabbitmq.client.Connection;
5. import com.rabbitmq.utils.ConnectionUntils;
6. /**
7. * 通配符模式
8. * 生产者
9. */
10. public class Producer {
11.     static final String EXCHANGE_NAME = "topic_exchange";
12.     static final String QUEUE_NAME1 = "topic_queue_1";
13.     static final String QUEUE_NAME2 = "topic_queue_2";
14.     public static void main(String[] args) throws Exception {
15.         Connection connection = ConnectionUntils.getConnection();
16.         Channel channel = connection.createChannel();
17.         channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
18.         channel.queueDeclare(QUEUE_NAME1, true, false, false, null);
19.         channel.queueDeclare(QUEUE_NAME2, true, false, false, null);
20.         channel.queueBind(QUEUE_NAME1,EXCHANGE_NAME,"item.*");
21.         channel.queueBind(QUEUE_NAME2,EXCHANGE_NAME,"item.insert");
22.         channel.queueBind(QUEUE_NAME2,EXCHANGE_NAME,"item.update");
23.         String message = "Hello World! - 新增 insert";
24.         channel.basicPublish(EXCHANGE_NAME,"item.insert",null,message.getBytes());
25.         System.out.println(" [x] Sent '" + message + "'");
26.         String message2 = "Hello World! - 修改 update";
27.         channel.basicPublish(EXCHANGE_NAME,"item.update",null,message2.getBytes());
28.         System.out.println(" [x] Sent '" + message2 + "'");
29.         String message3 = "Hello World! - 修改 delete";
30.         channel.basicPublish(EXCHANGE_NAME,"item.delete",null,message3.getBytes());
31.         System.out.println(" [x] Sent '" + message3 + "'");
32.         channel.close();
33.         connection.close();
34.     }
35. }

复制代码

1. package com.rabbitmq.topic;
2. import com.rabbitmq.client.*;
3. import com.rabbitmq.utils.ConnectionUntils;
4. import java.io.IOException;
5. public class Consumer1 {
6.     public static void main(String[] args) throws Exception {
7.         Connection conection = ConnectionUntils.getConnection();
8.         Channel channel = conection.createChannel();
9.         channel.queueDeclare(Producer.QUEUE_NAME1, true, false, false, null);
10.         channel.queueDeclare(Producer.QUEUE_NAME2, true, false, false, null);
11.         channel.queueBind(Producer.QUEUE_NAME1,Producer.EXCHANGE_NAME,"item.*");
12.         channel.queueBind(Producer.QUEUE_NAME2,Producer.EXCHANGE_NAME,"item.insert");
13.         channel.queueBind(Producer.QUEUE_NAME2,Producer.EXCHANGE_NAME,"item.update");
14.         DefaultConsumer consumer = new DefaultConsumer(channel) {
15.             @Override
16.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
17.                 System.out.println("路由：" + envelope.getRoutingKey());
18.                 System.out.println("交换机：" + envelope.getExchange());
19.                 System.out.println("消息ID：" + envelope.getDeliveryTag());
20.                 System.out.println("消息" + new String(body, "UTF-8"));
21.                 super.handleDelivery(consumerTag, envelope, properties, body);
22.             }
23.         };
24.         channel.basicConsume(Producer.QUEUE_NAME1, true, consumer);
25.     }
26. }

复制代码

1. package com.rabbitmq.topic;
2. import com.rabbitmq.client.*;
3. import com.rabbitmq.utils.ConnectionUntils;
4. import java.io.IOException;
5. public class Consumer2 {
6.     public static void main(String[] args) throws Exception {
7.         Connection conection = ConnectionUntils.getConnection();
8.         Channel channel = conection.createChannel();
9.         channel.queueDeclare(Producer.QUEUE_NAME1, true, false, false, null);
10.         channel.queueDeclare(Producer.QUEUE_NAME2, true, false, false, null);
11.         channel.queueBind(Producer.QUEUE_NAME1,Producer.EXCHANGE_NAME,"item.*");
12.         channel.queueBind(Producer.QUEUE_NAME2,Producer.EXCHANGE_NAME,"item.insert");
13.         channel.queueBind(Producer.QUEUE_NAME2,Producer.EXCHANGE_NAME,"item.update");
14.         DefaultConsumer consumer = new DefaultConsumer(channel) {
15.             @Override
16.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
17.                 System.out.println("路由：" + envelope.getRoutingKey());
18.                 System.out.println("交换机：" + envelope.getExchange());
19.                 System.out.println("消息ID：" + envelope.getDeliveryTag());
20.                 System.out.println("消息" + new String(body, "UTF-8"));
21.                 super.handleDelivery(consumerTag, envelope, properties, body);
22.             }
23.         };
24.         channel.basicConsume(Producer.QUEUE_NAME2, true, consumer);
25.     }
26. }

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。