rabbitmq介绍及rabbitmq在java中基础使用

叁歲伎倆 2022-05-25 09:24 165阅读 0赞

###### rabbitmq介绍及rabbitmq在java中基础使用 ######

https://blog.csdn.net/qq\_34021712/article/details/72567801

2017年05月20日 02:14:42

阅读数：1046

***原文地址，转载请注明出处：[http://blog.csdn.net/qq\_34021712/article/details/72567801][http_blog.csdn.net_qq_34021712_article_details_72567801]    ©[王赛超][Link 1]***

RabbitMQ简介  
RabbitMQ是一个在AMQP基础上完整的，可复用的企业消息系统。它可以用于大型软件系统各个模块之间的高效通信，支持高并发，支持可扩展。使用Erlang语言编写。  
  
RabbitMQ相关术语：  
          1.Broker：简单来说就是消息队列服务器实体。  
          2.Exchange：消息交换机，它指定消息按什么规则，路由到哪个队列。  
          3.Queue：消息队列载体，每个消息都会被投入到一个或多个队列。  
          4.Binding：绑定，它的作用就是把exchange和queue按照路由规则绑定起来。  
          5.Routing Key：路由关键字，exchange根据这个关键字进行消息投递。  
          6.vhost：虚拟主机，一个broker里可以开设多个vhost，用作不同用户的权限分离。  
          7.producer：消息生产者，就是投递消息的程序。  
          8.consumer：消息消费者，就是接受消息的程序。  
          9.channel：消息通道，在客户端的每个连接里，可建立多个channel，每个channel代表一个会话任务。  
![Center][]  
  
RabbitMQ常用发布订阅模式的运行流程：  
AMQP模型中，消息在producer中产生，发送到MQ的exchange上，exchange根据配置的路由方式发到相应的Queue上，Queue又将消息发送给consumer，消息从queue到consumer有push和pull两种方式。 消息队列的使用过程大概如下：  
  
          1.客户端连接到消息队列服务器，打开一个channel。  
          2.客户端声明一个exchange，并设置相关属性。  
          3.客户端声明一个queue，并设置相关属性。  
          4.客户端使用routing key，在exchange和queue之间建立好绑定关系。  
          5.客户端投递消息到exchange。  
  
exchange接收到消息后，就根据消息的key和已经设置的binding，进行消息路由，将消息投递到一个或多个队列里。 exchange也有几个类型，下面会有介绍。

RabbitMQ教程：

引入maven依赖:

**\[html\]** [view plain][] [copy][view plain]

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

1.hello world！

![Center 1][]

生产者:

**\[java\]** [view plain][] [copy][view plain]

1.  package com.rabbitmq.test.T\_helloworld;  
2.    
3.  import com.rabbitmq.client.Channel;  
4.  import com.rabbitmq.client.Connection;  
5.  import com.rabbitmq.test.util.ConnectionUtil;  
6.    
7.  /\*\* 
8.   \* helloworld 
9.   \* @author lenovo 
10.  \* 
11.  \*/  
12. public class Producer \{  
13.   
14.     private final static String QUEUE\_NAME = "test\_queue";  
15.   
16.     public static void main(String\[\] argv) throws Exception \{  
17.         // 获取到连接以及mq通道  
18.         Connection connection = ConnectionUtil.getConnection();  
19.         // 从连接中创建通道  
20.         Channel channel = connection.createChannel();  
21.   
22.         /\* 
23.          \* 声明（创建）队列 
24.          \* 参数1：队列名称 
25.          \* 参数2：为true时server重启队列不会消失 
26.          \* 参数3：队列是否是独占的，如果为true只能被一个connection使用，其他连接建立时会抛出异常 
27.          \* 参数4：队列不再使用时是否自动删除（没有连接，并且没有未处理的消息) 
28.          \* 参数5：建立队列时的其他参数 
29.          \*/  
30.         channel.queueDeclare(QUEUE\_NAME, false, false, false, null);  
31.   
32.         // 消息内容  
33.         String message = "Hello World!";  
34.         /\* 
35.          \* 向server发布一条消息 
36.          \* 参数1：exchange名字，若为空则使用默认的exchange 
37.          \* 参数2：routing key 
38.          \* 参数3：其他的属性 
39.          \* 参数4：消息体 
40.          \* RabbitMQ默认有一个exchange，叫default exchange，它用一个空字符串表示，它是direct exchange类型， 
41.          \* 任何发往这个exchange的消息都会被路由到routing key的名字对应的队列上，如果没有对应的队列，则消息会被丢弃 
42.          \*/  
43.         channel.basicPublish("", QUEUE\_NAME, null, message.getBytes());  
44.         System.out.println(" \[生产者\] Sent '" + message + "'");  
45.   
46.         //关闭通道和连接  
47.         channel.close();  
48.         connection.close();  
49.     \}  
50. \}

消费者:

**\[java\]** [view plain][] [copy][view plain]

1.  package com.rabbitmq.test.T\_helloworld;  
2.    
3.  import com.rabbitmq.client.Channel;  
4.  import com.rabbitmq.client.Connection;  
5.  import com.rabbitmq.client.QueueingConsumer;  
6.  import com.rabbitmq.test.util.ConnectionUtil;  
7.    
8.  public class Consumer \{  
9.    
10.     private final static String QUEUE\_NAME = "test\_queue";  
11.   
12.     public static void main(String\[\] argv) throws Exception \{  
13.   
14.         // 获取到连接以及mq通道  
15.         Connection connection = ConnectionUtil.getConnection();  
16.         // 从连接中创建通道  
17.         Channel channel = connection.createChannel();  
18.   
19.         // 声明队列(如果你已经明确的知道有这个队列,那么下面这句代码可以注释掉,如果不注释掉的话,也可以理解为消费者必须监听一个队列,如果没有就创建一个)  
20.         channel.queueDeclare(QUEUE\_NAME, false, false, false, null);  
21.   
22.         // 定义队列的消费者  
23.         QueueingConsumer consumer = new QueueingConsumer(channel);  
24.        /\* 
25.          \* 监听队列 
26.          \* 参数1:队列名称 
27.          \* 参数2：是否发送ack包，不发送ack消息会持续在服务端保存，直到收到ack。  可以通过channel.basicAck手动回复ack 
28.          \* 参数3：消费者 
29.          \*/   
30.         channel.basicConsume(QUEUE\_NAME, true, consumer);  
31.   
32.         // 获取消息  
33.         while (true) \{  
34.             QueueingConsumer.Delivery delivery = consumer.nextDelivery();  
35.             String message = new String(delivery.getBody());  
36.             System.out.println(" \[消费者\] Received '" + message + "'");  
37.         \}  
38.     \}  
39. \}

2.Work模式  
Work普通模式  
![Center 2][]

生产者:

**\[java\]** [view plain][] [copy][view plain]

1.  package com.rabbitmq.test.T\_work;  
2.    
3.  import com.rabbitmq.client.Channel;  
4.  import com.rabbitmq.client.Connection;  
5.  import com.rabbitmq.test.util.ConnectionUtil;  
6.    
7.  /\*\* 
8.   \* work模式 
9.   \* @author lenovo 
10.  \* 
11.  \*/  
12. public class Producer \{  
13.   
14.     private final static String QUEUE\_NAME = "test\_queue\_work";  
15.   
16.     public static void main(String\[\] argv) throws Exception \{  
17.         // 获取到连接以及mq通道  
18.         Connection connection = ConnectionUtil.getConnection();  
19.         Channel channel = connection.createChannel();  
20.   
21.         // 声明队列  
22.         channel.queueDeclare(QUEUE\_NAME, false, false, false, null);  
23.   
24.         for (int i = 0; i < 50; i++) \{  
25.             // 消息内容  
26.             String message = "" + i;  
27.             channel.basicPublish("", QUEUE\_NAME, null, message.getBytes());  
28.             System.out.println(" \[生产者\] Sent '" + message + "'");  
29.             //发送的消息间隔越来越长  
30.             Thread.sleep(i \* 10);  
31.         \}  
32.   
33.         channel.close();  
34.         connection.close();  
35.     \}  
36. \}

消费者1:

**\[java\]** [view plain][] [copy][view plain]

1.  package com.rabbitmq.test.T\_work;  
2.    
3.  import com.rabbitmq.client.Channel;  
4.  import com.rabbitmq.client.Connection;  
5.  import com.rabbitmq.client.QueueingConsumer;  
6.  import com.rabbitmq.test.util.ConnectionUtil;  
7.    
8.  public class Consumer1 \{  
9.    
10.     private final static String QUEUE\_NAME = "test\_queue\_work";  
11.   
12.     public static void main(String\[\] argv) throws Exception \{  
13.   
14.         // 获取到连接以及mq通道  
15.         Connection connection = ConnectionUtil.getConnection();  
16.         Channel channel = connection.createChannel();  
17.   
18.         // 声明队列  
19.         channel.queueDeclare(QUEUE\_NAME, false, false, false, null);  
20.           
21.         // 同一时刻服务器只会发一条消息给消费者(能者多劳模式)  
22.         //channel.basicQos(1);  
23.   
24.         // 定义队列的消费者  
25.         QueueingConsumer consumer = new QueueingConsumer(channel);  
26.         /\* 
27.          \* 监听队列，不自动返回ack包,下面手动返回 
28.          \* 如果不回复，消息不会在服务器删除 
29.          \*/   
30.         channel.basicConsume(QUEUE\_NAME, false, consumer);  
31.   
32.         // 获取消息  
33.         while (true) \{  
34.             QueueingConsumer.Delivery delivery = consumer.nextDelivery();  
35.             String message = new String(delivery.getBody());  
36.             System.out.println(" \[消费者1\] Received '" + message + "'");  
37.             //休眠  
38.             Thread.sleep(10);  
39.             // 手动返回ack包确认状态  
40.             channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);  
41.           //channel.basicReject(); channel.basicNack(); //可以通过这两个函数拒绝消息，可以指定消息在服务器删除还是继续投递给其他消费者  
42.         \}  
43.     \}  
44. \}

消费者2:

**\[java\]** [view plain][] [copy][view plain]

1.  package com.rabbitmq.test.T\_work;  
2.    
3.  import com.rabbitmq.client.Channel;  
4.  import com.rabbitmq.client.Connection;  
5.  import com.rabbitmq.client.QueueingConsumer;  
6.  import com.rabbitmq.test.util.ConnectionUtil;  
7.    
8.  public class Consumer2 \{  
9.    
10.     private final static String QUEUE\_NAME = "test\_queue\_work";  
11.   
12.     public static void main(String\[\] argv) throws Exception \{  
13.   
14.         // 获取到连接以及mq通道  
15.         Connection connection = ConnectionUtil.getConnection();  
16.         Channel channel = connection.createChannel();  
17.   
18.         // 声明队列  
19.         channel.queueDeclare(QUEUE\_NAME, false, false, false, null);  
20.   
21.         // 同一时刻服务器只会发一条消息给消费者(能者多劳模式)  
22.         //channel.basicQos(1);  
23.   
24.         // 定义队列的消费者  
25.         QueueingConsumer consumer = new QueueingConsumer(channel);  
26.         // 监听队列，手动返回完成状态  
27.         channel.basicConsume(QUEUE\_NAME, false, consumer);  
28.   
29.         // 获取消息  
30.         while (true) \{  
31.             QueueingConsumer.Delivery delivery = consumer.nextDelivery();  
32.             String message = new String(delivery.getBody());  
33.             System.out.println(" \[消费者2\] Received '" + message + "'");  
34.             // 休眠1秒  
35.             Thread.sleep(1000);  
36.   
37.             //反馈消息的消费状态  
38.             channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);  
39.         \}  
40.     \}  
41. \}

测试结果：  
1、消费者1和消费者2获取到的消息内容是不同的，同一个消息只能被一个消费者获取。  
2、消费者1和消费者2获取到的消息的数量是相同的，一个是奇数一个是偶数。  
  
其实，这样是不合理的，应该是消费者1要比消费者2获取到的消息多才对。  
Work的能者多劳模式  
需要将上面两个消费者的 channel.basicQos(1);这行代码的注释打开,再次执行会发现,休眠时间短的消费者执行的任务多  
消息的确认  
在以上的代码中,已经给出了注释,如何使用自动确认和手动确认,消费者从队列中获取消息，服务端如何知道消息已经被消费呢？  
模式1：自动确认  
只要消息从队列中获取，无论消费者获取到消息后是否成功消息，都认为是消息已经成功消费。  
模式2：手动确认  
消费者从队列中获取消息后，服务器会将该消息标记为不可用状态，等待消费者的反馈，如果消费者一直没有反馈，那么该消息将一直处于不可用状态。  
如果选用自动确认,在消费者拿走消息执行过程中出现宕机时,消息可能就会丢失！！  
![Center 3][]

3.订阅模式

![Center 4][]  
  
文章开头有发布订阅的流程介绍

生产者：

**\[java\]** [view plain][] [copy][view plain]

1.  package com.rabbitmq.test.T\_pubsub;  
2.    
3.  import com.rabbitmq.client.Channel;  
4.  import com.rabbitmq.client.Connection;  
5.  import com.rabbitmq.test.util.ConnectionUtil;  
6.    
7.  /\*\* 
8.   \* 订阅模式 
9.   \* @author lenovo 
10.  \* 
11.  \*/  
12. public class Producer \{  
13.   
14.     //交换机的名称  
15.     private final static String EXCHANGE\_NAME = "test\_exchange\_fanout";  
16.   
17.     public static void main(String\[\] argv) throws Exception \{  
18.         // 获取到连接以及mq通道  
19.         Connection connection = ConnectionUtil.getConnection();  
20.         Channel channel = connection.createChannel();  
21.   
22.         /\* 
23.          \* 声明exchange(交换机) 
24.          \* 参数1：交换机名称 
25.          \* 参数2：交换机类型 
26.          \* 参数3：交换机持久性，如果为true则服务器重启时不会丢失 
27.          \* 参数4：交换机在不被使用时是否删除 
28.          \* 参数5：交换机的其他属性 
29.          \*/  
30.         channel.exchangeDeclare(EXCHANGE\_NAME, "fanout",true,true,null);  
31.   
32.         // 消息内容  
33.         String message = "订阅消息";  
34.         channel.basicPublish(EXCHANGE\_NAME, "", null, message.getBytes());  
35.         System.out.println(" \[生产者\] Sent '" + message + "'");  
36.   
37.         channel.close();  
38.         connection.close();  
39.     \}  
40. \}

消费者1：

**\[java\]** [view plain][] [copy][view plain]

1.  package com.rabbitmq.test.T\_pubsub;  
2.    
3.  import com.rabbitmq.client.Channel;  
4.  import com.rabbitmq.client.Connection;  
5.  import com.rabbitmq.client.QueueingConsumer;  
6.  import com.rabbitmq.test.util.ConnectionUtil;  
7.    
8.  public class Consumer1 \{  
9.    
10.     private final static String QUEUE\_NAME = "test\_queue\_exchange\_1";  
11.   
12.     private final static String EXCHANGE\_NAME = "test\_exchange\_fanout";  
13.   
14.     public static void main(String\[\] argv) throws Exception \{  
15.   
16.         // 获取到连接以及mq通道  
17.         Connection connection = ConnectionUtil.getConnection();  
18.         Channel channel = connection.createChannel();  
19.   
20.         // 声明队列  
21.         channel.queueDeclare(QUEUE\_NAME, false, false, false, null);  
22.   
23.           
24.         /\* 
25.          \* 绑定队列到交换机(这个交换机的名称一定要和上面的生产者交换机名称相同) 
26.          \* 参数1：队列的名称 
27.          \* 参数2：交换机的名称 
28.          \* 参数3：Routing Key 
29.          \*  
30.          \*/  
31.         channel.queueBind(QUEUE\_NAME, EXCHANGE\_NAME, "");  
32.   
33.         // 同一时刻服务器只会发一条消息给消费者  
34.         channel.basicQos(1);  
35.   
36.         // 定义队列的消费者  
37.         QueueingConsumer consumer = new QueueingConsumer(channel);  
38.         // 监听队列，手动返回完成  
39.         channel.basicConsume(QUEUE\_NAME, false, consumer);  
40.   
41.         // 获取消息  
42.         while (true) \{  
43.             QueueingConsumer.Delivery delivery = consumer.nextDelivery();  
44.             String message = new String(delivery.getBody());  
45.             System.out.println(" \[消费者1\] Received '" + message + "'");  
46.             Thread.sleep(10);  
47.               
48.             channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);  
49.         \}  
50.     \}  
51. \}

消费者2：

**\[java\]** [view plain][] [copy][view plain]

1.  package com.rabbitmq.test.T\_pubsub;  
2.    
3.  import com.rabbitmq.client.Channel;  
4.  import com.rabbitmq.client.Connection;  
5.  import com.rabbitmq.client.QueueingConsumer;  
6.  import com.rabbitmq.test.util.ConnectionUtil;  
7.    
8.  public class Consumer2 \{  
9.    
10.     private final static String QUEUE\_NAME = "test\_queue\_exchange\_2";  
11.   
12.     private final static String EXCHANGE\_NAME = "test\_exchange\_fanout";  
13.   
14.     public static void main(String\[\] argv) throws Exception \{  
15.   
16.         // 获取到连接以及mq通道  
17.         Connection connection = ConnectionUtil.getConnection();  
18.         Channel channel = connection.createChannel();  
19.   
20.         // 声明队列  
21.         channel.queueDeclare(QUEUE\_NAME, false, false, false, null);  
22.   
23.         // 绑定队列到交换机  
24.         channel.queueBind(QUEUE\_NAME, EXCHANGE\_NAME, "");  
25.   
26.         // 同一时刻服务器只会发一条消息给消费者  
27.         channel.basicQos(1);  
28.   
29.         // 定义队列的消费者  
30.         QueueingConsumer consumer = new QueueingConsumer(channel);  
31.         // 监听队列，手动返回完成  
32.         channel.basicConsume(QUEUE\_NAME, false, consumer);  
33.   
34.         // 获取消息  
35.         while (true) \{  
36.             QueueingConsumer.Delivery delivery = consumer.nextDelivery();  
37.             String message = new String(delivery.getBody());  
38.             System.out.println(" \[消费者2\] Received '" + message + "'");  
39.             Thread.sleep(10);  
40.   
41.             channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);  
42.         \}  
43.     \}  
44. \}

注意：消息发送到没有队列绑定的交换机时，消息将丢失，因为，交换机没有存储消息的能力，消息只能存在在队列中。  
  
  
Exchange类型  
Direct 、Fanout 、Topic  三种类型，RabbitMQ默认有一个exchange，叫default exchange，它用一个空字符串表示，它是direct exchange类型。  
![Center 5][]

下面介绍的路由模式和通配符模式都是属于订阅模式,只不过加入了Routing Key(路由键,文章开头有介绍)。

3.1路由模式

![Center 6][]

生产者：

**\[java\]** [view plain][] [copy][view plain]

1.  package com.rabbitmq.test.T\_routing;  
2.    
3.  import com.rabbitmq.client.Channel;  
4.  import com.rabbitmq.client.Connection;  
5.  import com.rabbitmq.test.util.ConnectionUtil;  
6.    
7.  /\*\* 
8.   \* 路由模式 
9.   \* @author lenovo 
10.  \* 
11.  \*/  
12. public class Producer \{  
13.   
14.     private final static String EXCHANGE\_NAME = "test\_exchange\_direct";  
15.   
16.     public static void main(String\[\] argv) throws Exception \{  
17.         // 获取到连接以及mq通道  
18.         Connection connection = ConnectionUtil.getConnection();  
19.         Channel channel = connection.createChannel();  
20.   
21.         // 声明exchange  
22.         channel.exchangeDeclare(EXCHANGE\_NAME, "direct");  
23.   
24.         // 消息内容  
25.         String message = "这是消息B";  
26.         channel.basicPublish(EXCHANGE\_NAME, "B", null, message.getBytes());  
27.         System.out.println(" \[生产者\] Sent '" + message + "'");  
28.   
29.         channel.close();  
30.         connection.close();  
31.     \}  
32. \}

消费者1：

**\[java\]** [view plain][] [copy][view plain]

1.  package com.rabbitmq.test.T\_routing;  
2.    
3.  import com.rabbitmq.client.Channel;  
4.  import com.rabbitmq.client.Connection;  
5.  import com.rabbitmq.client.QueueingConsumer;  
6.  import com.rabbitmq.test.util.ConnectionUtil;  
7.    
8.  public class Consumer1 \{  
9.    
10.     private final static String QUEUE\_NAME = "test\_queue\_direct\_1";  
11.   
12.     private final static String EXCHANGE\_NAME = "test\_exchange\_direct";  
13.   
14.     public static void main(String\[\] argv) throws Exception \{  
15.   
16.         // 获取到连接以及mq通道  
17.         Connection connection = ConnectionUtil.getConnection();  
18.         Channel channel = connection.createChannel();  
19.   
20.         // 声明队列  
21.         channel.queueDeclare(QUEUE\_NAME, false, false, false, null);  
22.   
23.         /\* 
24.          \* 绑定队列到交换机 
25.          \* 参数1：队列的名称 
26.          \* 参数2：交换机的名称 
27.          \* 参数3：routingKey 
28.          \*/  
29.         channel.queueBind(QUEUE\_NAME, EXCHANGE\_NAME, "A");  
30.   
31.         // 同一时刻服务器只会发一条消息给消费者  
32.         channel.basicQos(1);  
33.   
34.         // 定义队列的消费者  
35.         QueueingConsumer consumer = new QueueingConsumer(channel);  
36.         // 监听队列，手动返回完成  
37.         channel.basicConsume(QUEUE\_NAME, false, consumer);  
38.   
39.         // 获取消息  
40.         while (true) \{  
41.             QueueingConsumer.Delivery delivery = consumer.nextDelivery();  
42.             String message = new String(delivery.getBody());  
43.             System.out.println(" \[消费者1\] Received '" + message + "'");  
44.             Thread.sleep(10);  
45.   
46.             channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);  
47.         \}  
48.     \}  
49. \}

消费者2：

**\[java\]** [view plain][] [copy][view plain]

1.  package com.rabbitmq.test.T\_routing;  
2.    
3.  import com.rabbitmq.client.Channel;  
4.  import com.rabbitmq.client.Connection;  
5.  import com.rabbitmq.client.QueueingConsumer;  
6.  import com.rabbitmq.test.util.ConnectionUtil;  
7.    
8.  public class Consumer2 \{  
9.    
10.     private final static String QUEUE\_NAME = "test\_queue\_direct\_2";  
11.   
12.     private final static String EXCHANGE\_NAME = "test\_exchange\_direct";  
13.   
14.     public static void main(String\[\] argv) throws Exception \{  
15.   
16.         // 获取到连接以及mq通道  
17.         Connection connection = ConnectionUtil.getConnection();  
18.         Channel channel = connection.createChannel();  
19.   
20.         // 声明队列  
21.         channel.queueDeclare(QUEUE\_NAME, false, false, false, null);  
22.   
23.         // 绑定队列到交换机  
24.         channel.queueBind(QUEUE\_NAME, EXCHANGE\_NAME, "B");  
25.         //如果想让消费者2同时接受routingKey为A 和为B的消息,只要在下面在此添加一个Bing就可以了  
26.         channel.queueBind(QUEUE\_NAME, EXCHANGE\_NAME, "A");  
27.   
28.         // 同一时刻服务器只会发一条消息给消费者  
29.         channel.basicQos(1);  
30.   
31.         // 定义队列的消费者  
32.         QueueingConsumer consumer = new QueueingConsumer(channel);  
33.         // 监听队列，手动返回完成  
34.         channel.basicConsume(QUEUE\_NAME, false, consumer);  
35.   
36.         // 获取消息  
37.         while (true) \{  
38.             QueueingConsumer.Delivery delivery = consumer.nextDelivery();  
39.             String message = new String(delivery.getBody());  
40.             System.out.println(" \[消费者2\] Received '" + message + "'");  
41.             Thread.sleep(10);  
42.   
43.             channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);  
44.         \}  
45.     \}  
46. \}

3.2通配符模式  
将路由键和某模式进行匹配。此时队列需要绑定要一个模式上。符号“\#”匹配一个或多个词，符号“\*”只能匹配一个词。因此“audit.\#”能够匹配到“audit.irs”和“audit.irs.corporate”，但是“audit.\*” 只会匹配到“audit.irs”。  
![Center 7][] ![Center 8][]

生产者：

**\[java\]** [view plain][] [copy][view plain]

1.  package com.rabbitmq.test.T\_topic;  
2.    
3.  import com.rabbitmq.client.Channel;  
4.  import com.rabbitmq.client.Connection;  
5.  import com.rabbitmq.test.util.ConnectionUtil;  
6.    
7.  /\*\* 
8.   \* 通配模式 
9.   \* @author lenovo 
10.  \* 
11.  \*/  
12. public class Producer \{  
13.   
14.     private final static String EXCHANGE\_NAME = "test\_exchange\_topic";  
15.   
16.     public static void main(String\[\] argv) throws Exception \{  
17.         // 获取到连接以及mq通道  
18.         Connection connection = ConnectionUtil.getConnection();  
19.         Channel channel = connection.createChannel();  
20.   
21.         // 声明exchange  
22.         channel.exchangeDeclare(EXCHANGE\_NAME, "topic");  
23.   
24.         // 消息内容  模拟 有人购物下订单  
25.         String message = "新增订单:id=101";  
26.         channel.basicPublish(EXCHANGE\_NAME, "order.insert", null, message.getBytes());  
27.         System.out.println(" \[生产者\] Sent '" + message + "'");  
28.   
29.         channel.close();  
30.         connection.close();  
31.     \}  
32.       
33. \}

消费者1：

**\[java\]** [view plain][] [copy][view plain]

1.  package com.rabbitmq.test.T\_topic;  
2.    
3.  import com.rabbitmq.client.Channel;  
4.  import com.rabbitmq.client.Connection;  
5.  import com.rabbitmq.client.QueueingConsumer;  
6.  import com.rabbitmq.test.util.ConnectionUtil;  
7.    
8.  public class Consumer1 \{  
9.    
10.     private final static String QUEUE\_NAME = "test\_queue\_topic\_1";  
11.   
12.     private final static String EXCHANGE\_NAME = "test\_exchange\_topic";  
13.   
14.     public static void main(String\[\] argv) throws Exception \{  
15.   
16.         // 获取到连接以及mq通道  
17.         Connection connection = ConnectionUtil.getConnection();  
18.         Channel channel = connection.createChannel();  
19.   
20.         // 声明队列  
21.         channel.queueDeclare(QUEUE\_NAME, false, false, false, null);  
22.   
23.         // 绑定队列到交换机  
24.         channel.queueBind(QUEUE\_NAME, EXCHANGE\_NAME, "order.\#");  
25.   
26.         // 同一时刻服务器只会发一条消息给消费者  
27.         channel.basicQos(1);  
28.   
29.         // 定义队列的消费者  
30.         QueueingConsumer consumer = new QueueingConsumer(channel);  
31.         // 监听队列，手动返回完成  
32.         channel.basicConsume(QUEUE\_NAME, false, consumer);  
33.   
34.         // 获取消息  
35.         while (true) \{  
36.             QueueingConsumer.Delivery delivery = consumer.nextDelivery();  
37.             String message = new String(delivery.getBody());  
38.             System.out.println(" \[财务系统\] Received '" + message + "'");  
39.             Thread.sleep(10);  
40.   
41.             channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);  
42.         \}  
43.     \}  
44. \}

消费者2：

**\[java\]** [view plain][] [copy][view plain]

1.  package com.rabbitmq.test.T\_topic;  
2.    
3.  import com.rabbitmq.client.Channel;  
4.  import com.rabbitmq.client.Connection;  
5.  import com.rabbitmq.client.QueueingConsumer;  
6.  import com.rabbitmq.test.util.ConnectionUtil;  
7.    
8.  public class Consumer2 \{  
9.    
10.     private final static String QUEUE\_NAME = "test\_queue\_topic\_2";  
11.   
12.     private final static String EXCHANGE\_NAME = "test\_exchange\_topic";  
13.   
14.     public static void main(String\[\] argv) throws Exception \{  
15.   
16.         // 获取到连接以及mq通道  
17.         Connection connection = ConnectionUtil.getConnection();  
18.         Channel channel = connection.createChannel();  
19.   
20.         // 声明队列  
21.         channel.queueDeclare(QUEUE\_NAME, false, false, false, null);  
22.   
23.         // 绑定队列到交换机  
24.         channel.queueBind(QUEUE\_NAME, EXCHANGE\_NAME, "order.insert");  
25.   
26.         // 同一时刻服务器只会发一条消息给消费者  
27.         channel.basicQos(1);  
28.   
29.         // 定义队列的消费者  
30.         QueueingConsumer consumer = new QueueingConsumer(channel);  
31.         // 监听队列，手动返回完成  
32.         channel.basicConsume(QUEUE\_NAME, false, consumer);  
33.   
34.         // 获取消息  
35.         while (true) \{  
36.             QueueingConsumer.Delivery delivery = consumer.nextDelivery();  
37.             String message = new String(delivery.getBody());  
38.             System.out.println(" \[物流系统\] Received '" + message + "'");  
39.             Thread.sleep(10);  
40.   
41.             channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);  
42.         \}  
43.     \}  
44. \}

个人分类： [rabbitmq ][rabbitmq]

[http_blog.csdn.net_qq_34021712_article_details_72567801]: http://blog.csdn.net/qq_34021712/article/details/72567801
[Link 1]: http://blog.csdn.net/qq_34021712
[Center]: /images/20220525/cf5f477ebc6e45d8b86d50fcb777c713.png
[view plain]: https://blog.csdn.net/qq_34021712/article/details/72567801#
[Center 1]: /images/20220525/0e611077c8744959a9d1fa5945facbe8.png
[Center 2]: /images/20220525/7caf2b3de97d4f4c9b4b3c2690b3c276.png
[Center 3]: /images/20220525/204b07a592e440f2ac7ffb3588625ac1.png
[Center 4]: /images/20220525/90f665d768254f7a87ee0c80fae2e1c5.png
[Center 5]: /images/20220525/28ddf4fbce8f486e9ffe21ebc1b55456.png
[Center 6]: /images/20220525/43931cd90c004de78f0b81b3ef724c0d.png
[Center 7]: https://img-blog.csdn.net/20170520024814651?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvcXFfMzQwMjE3MTI=/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/Center
[Center 8]: /images/20220525/3462be125a8f48a6822a248de3dba291.png
[rabbitmq]: https://blog.csdn.net/qq_34021712/article/category/6928540