RabbitMQ的五种消息模子

相干RabbitMQ学习链接：
  1.初学RabbitMQ-CSDN博客
  2.安装RabbitMQ-CSDN博客
  3.SpringBoot整合RabbitMQ-CSDN博客
  RabbitMQ提供了多种消息模子，官网上第6种是RPC不属于常规的消息队列。 属于消息模子的是前5种：

• 简朴的一对一模子
  
• 工作队列模子 ，一个生产者将消息分发给多个消费者
  
• 发布/订阅模子 ，生产者发布消息，多个消费者同时收取
  
• 路由模子 ，生产者通过关键字发送消息给特定消费者
  
• 主题模子 ，路由模式基础上，在关键字里参加了通配符
  

 一对一模子

最根本的队列模子： 一个生产者发送消息到一个队列，一个消费者从队列中取消息。
操作步骤

1）启动Rabbitmq，在管理页面中创建用户admin 2）使用admin登录，然后创建假造主机myhost

创建队列，设置如下

案例代码

导入依赖

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

复制代码

开辟工具类

1. public class MQUtils {
3.     public static final String QUEUE_NAME = "myqueue01";
4.     public static final String QUEUE_NAME2 = "myqueue02";
6.     public static final String EXCHANGE_NAME = "myexchange01";
7.     public static final String EXCHANGE_NAME2 = "myexchange02";
8.     public static final String EXCHANGE_NAME3 = "myexchange03";
10.     /**
11.      * 获得MQ的连接
12.      * @return
13.      * @throws IOException
14.      */
15.     public static Connection getConnection() throws IOException {
16.         ConnectionFactory connectionFactory = new ConnectionFactory();
17.         //配置服务器名、端口、虚拟主机名、登录账号和密码
18.         connectionFactory.setHost("localhost");
19.         connectionFactory.setPort(5672);
20.         connectionFactory.setVirtualHost("myhost");
21.         connectionFactory.setUsername("admin");
22.         connectionFactory.setPassword("123456");
23.         return connectionFactory.newConnection();
24.     }
25. }

复制代码

开辟生产者

1. /**
2. * 生产者，发送简单的消息到队列中
3. */
4. public class SimpleProducer {
6.     public static void main(String[] args) throws IOException {
7.         Connection connection = MQUtils.getConnection();
8.         //创建通道
9.         Channel channel = connection.createChannel();
10.         //定义队列
11.         channel.queueDeclare(MQUtils.QUEUE_NAME,false,false,false,null);
12.         String msg = "Hello World!";
13.         //发布消息到队列
14.         channel.basicPublish("",MQUtils.QUEUE_NAME,null,msg.getBytes());
15.         channel.close();
16.         connection.close();
17.     }
18. }

复制代码

运行生产者代码，管理页面点进myqueue01，在GetMessages中可以看到消息

开辟消费者

1. /**
2. * 消费者，从队列中读取简单的消息
3. */
4. public class SimpleConsumer {
6.     public static void main(String[] args) throws IOException, InterruptedException {
7.         Connection connection = MQUtils.getConnection();
8.         Channel channel = connection.createChannel();
9.         //定义队列
10.         channel.queueDeclare(MQUtils.QUEUE_NAME,false,false,false,null);
11.         //创建消费者
12.         QueueingConsumer queueingConsumer = new QueueingConsumer(channel);
13.         //消费者消费通道中的消息
14.         channel.basicConsume(MQUtils.QUEUE_NAME,true,queueingConsumer);
15.         //读取消息
16.         while(true){
17.             QueueingConsumer.Delivery delivery = queueingConsumer.nextDelivery();
18.             System.out.println(new String(delivery.getBody()));
19.         }
20.     }
21. }

复制代码

工作队列模子

工作队列，生产者将消息分发给多个消费者，假如生产者生产了100条消息，消费者1消费50条，消费者2消费50条。
案例代码

开辟生产者

1. /**
2.   多对多模式的生产者，会发送多条消息到队列中
3. */
4. public class WorkProductor {
6.     public static void main(String[] args) throws IOException, InterruptedException {
7.         Connection connection = MQUtils.getConnection();
8.         Channel channel = connection.createChannel();
9.         channel.queueDeclare(MQUtils.QUEUE_NAME,false,false,false,null);
10.         for(int i = 0;i < 100;i++){
11.             String msg = "Hello-->" + i;
12.             channel.basicPublish("",MQUtils.QUEUE_NAME,null, msg.getBytes());
13.             System.out.println("send:" + msg);
14.             Thread.sleep(10);
15.         }
16.         channel.close();
17.         connection.close();
18.     }
19. }

复制代码

开辟消费者1

1. /**
2. * 多对多模式的消费者1
3. */
4. public class WorkConsumer01 {
6.     public static void main(String[] args) throws IOException, InterruptedException {
7.         Connection connection = MQUtils.getConnection();
8.         Channel channel = connection.createChannel();
9.         channel.queueDeclare(MQUtils.QUEUE_NAME,false,false,false,null);
10.         QueueingConsumer queueingConsumer = new QueueingConsumer(channel);
11.         //消费者消费通道中的消息
12.         channel.basicConsume(MQUtils.QUEUE_NAME,true,queueingConsumer);
13.         while(true){
14.             QueueingConsumer.Delivery delivery = queueingConsumer.nextDelivery();
15.             System.out.println("WorkConsumer1 receive :" + new String(delivery.getBody()));
16.             Thread.sleep(10);
17.         }
18.     }
19. }

复制代码

开辟消费者2

1. /**
2. * 多对多模式的消费者2
3. */
4. public class WorkConsumer02 {
6.     public static void main(String[] args) throws IOException, InterruptedException {
7.         Connection connection = MQUtils.getConnection();
8.         Channel channel = connection.createChannel();
9.         channel.queueDeclare(MQUtils.QUEUE_NAME,false,false,false,null);
10.         QueueingConsumer queueingConsumer = new QueueingConsumer(channel);
11.         //消费者消费通道中的消息
12.         channel.basicConsume(MQUtils.QUEUE_NAME,true,queueingConsumer);
13.         while(true){
14.             QueueingConsumer.Delivery delivery = queueingConsumer.nextDelivery();
15.             System.out.println("WorkConsumer2 receive :" + new String(delivery.getBody()));
16.             Thread.sleep(1000);
17.         }
18.     }
19. }

复制代码

生产者发送100个消息，两个消费者分别读取了50条。 看消息内容，发现队列发送消息采用的是轮询方式，也就是先发给消费者1，再发给消费者2，依次往复。
能者多劳

上面案例中有一个问题：消费者处理消息的速度是不一样的，消费者1处理后睡眠10毫秒（Thread.sleep(10)），消费者2是1000毫秒，速度相差100倍，但是末了处理的消息数还是一样的。这样就存在服从问题：处理能力强的消费者得不到更多的消息。
因为队列默认采用是主动确认机制，消息发过去后就主动确认，队列不清楚每个消息详细什么时间处理完，以是平均分配消息数量。
实现能者多劳：

• channel.basicQos(1);限制队列一次发一个消息给消费者，等消费者有了反馈，再发下一条
  
• channel.basicAck 消费完消息后手动反馈，处理快的消费者就能处理更多消息
  
• basicConsume 中的参数改为false
  

1. /**
2. 多对多模式的消费者1
3. */
4. public class WorkConsumer1 {
6.    public static void main(String[] args) throws IOException, InterruptedException {
7.        Connection connection = MQUtils.getConnection();
8.        Channel channel = connection.createChannel();
9.        channel.queueDeclare(MQUtils.QUEUE_NAME,false,false,false,null);
10.        //同一时刻服务器只发送一条消息给消费者
11.        channel.basicQos(1);
12.        QueueingConsumer queueingConsumer = new QueueingConsumer(channel);
13.        //true是自动返回完成状态，false表示手动
14.        channel.basicConsume(MQUtils.QUEUE_NAME,false,queueingConsumer);
15.        while(true){
16.            QueueingConsumer.Delivery delivery = queueingConsumer.nextDelivery();
17.            System.out.println("WorkConsumer1 receive :" + new String(delivery.getBody()));
18.            Thread.sleep(10);
19.            //手动确定返回状态，不写就是自动确认
20.            channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);
21.        }
22.    }
23. }
24.    
25. /**
26. * 多对多模式的消费者2
27. */
28. public class WorkConsumer2 {
30.    public static void main(String[] args) throws IOException, InterruptedException {
31.        Connection connection = MQUtils.getConnection();
32.        Channel channel = connection.createChannel();
33.        channel.queueDeclare(MQUtils.QUEUE_NAME,false,false,false,null);
34.        //同一时刻服务器只发送一条消息给消费者
35.        channel.basicQos(1);
36.        QueueingConsumer queueingConsumer = new QueueingConsumer(channel);
37.        //true是自动返回完成状态，false表示手动
38.        channel.basicConsume(MQUtils.QUEUE_NAME,false,queueingConsumer);
39.        while(true){
40.            QueueingConsumer.Delivery delivery = queueingConsumer.nextDelivery();
41.            System.out.println("WorkConsumer2 receive :" + new String(delivery.getBody()));
42.            Thread.sleep(1000);
43.            //手动确定返回状态，不写就是自动确认
44.            channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);
45.        }
46.    }
47. }

复制代码

发布/订阅模子

发布/订阅模式和Work模式的区别是：Work模式只存在一个队列，多个消费者共同消费一个队列中的消息；而发布订阅模式存在多个队列，不同的消费者可以从各自的队列中处理完全相同的消息。
 操作步骤

实现步骤： 1) 创建交换机（Exchange）范例是fanout（扇出） 2) 交换机必要绑定不同的队列 3) 不同的消费者从不同的队列中得到消息 4) 生产者发送消息到交换机 5) 再由交换机将消息分发到多个队列
新建队列

新建交换机

点击交换机，在bindings里面绑定两个队列

案例代码

生产者

1. /**
2. * 发布和订阅模式的生产者，消息会通过交换机发到队列
3. */
4. public class PublishProductor {
6.     public static void main(String[] args) throws IOException {
7.         Connection connection = MQUtils.getConnection();
8.         Channel channel = connection.createChannel();
9.         //声明fanout exchange
10.         channel.exchangeDeclare(MQUtils.EXCHANGE_NAME,"fanout");
11.         String msg = "Hello Fanout";
12.         //发布消息到交换机
13.         channel.basicPublish(MQUtils.EXCHANGE_NAME,"",null,msg.getBytes());
14.         System.out.println("send:" + msg);
15.         channel.close();
16.         connection.close();
17.     }
18. }

复制代码

消费者1

1. [/code] [code]/**
2. * 发布订阅模式的消费者1
3. * 两个消费者绑定的消息队列不同
4. * 通过交换机一个消息能被不同队列的两个消费者同时获取
5. * 一个队列可以有多个消费者，队列中的消息只能被一个消费者获取
6. */
7. public class SubscribeConsumer1 {
9.     public static void main(String[] args) throws IOException, InterruptedException {
10.         Connection connection = MQUtils.getConnection();
11.         Channel channel = connection.createChannel();
12.         channel.queueDeclare(MQUtils.QUEUE_NAME,false,false,false,null);
13.         //绑定队列1到交换机
14.         channel.queueBind(MQUtils.QUEUE_NAME,MQUtils.EXCHANGE_NAME,"");
15.         QueueingConsumer queueingConsumer = new QueueingConsumer(channel);
16.         channel.basicConsume(MQUtils.QUEUE_NAME,true,queueingConsumer);
17.         while(true){
18.             QueueingConsumer.Delivery delivery = queueingConsumer.nextDelivery();
19.             System.out.println("Consumer1 receive :" + new String(delivery.getBody()));
20.         }
21.     }
22. }

复制代码

消费者2

1. public class SubscribeConsumer2 {
3.     public static void main(String[] args) throws IOException, InterruptedException {
4.         Connection connection = MQUtils.getConnection();
5.         Channel channel = connection.createChannel();
6.         channel.queueDeclare(MQUtils.QUEUE_NAME2,false,false,false,null);
7.         //绑定队列2到交换机
8.         channel.queueBind(MQUtils.QUEUE_NAME2,MQUtils.EXCHANGE_NAME,"");
9.         QueueingConsumer queueingConsumer = new QueueingConsumer(channel);
10.         channel.basicConsume(MQUtils.QUEUE_NAME2,true,queueingConsumer);
11.         while(true){
12.             QueueingConsumer.Delivery delivery = queueingConsumer.nextDelivery();
13.             System.out.println("Consumer2 receive :" + new String(delivery.getBody()));
14.         }
15.     }
16. }

复制代码

路由模子

路由模式的消息队列可以给队列绑定不同的key，生产者发送消息时，给消息设置不同的key，这样交换机在分发消息时，可以让消息路由到key匹配的队列中。 可以想象上图是一个日记处理系统，C1可以处理error日记消息，C2可以处理info\error\warining范例的日记消息，使用路由模式就很轻易实现了。
操作步骤

新建direct范例的交换机

案例代码

生产者，给myqueue01绑定了key:error，myqueue02绑定了key:debug，然后发送了key:error的消息

1. /**
2.   路由模式的生产者，发布消息会有特定的Key，消息会被绑定特定Key的消费者获取
3. */
4. public class RouteProductor {
6.     public static void main(String[] args) throws IOException, InterruptedException {
7.         Connection connection = MQUtils.getConnection();
8.         Channel channel = connection.createChannel();
9.         //声明交换机类型为direct
10.         channel.exchangeDeclare(MQUtils.EXCHANGE_NAME2,"direct");
11.         String msg = "Hello-->Route";
12.         //绑定队列1到交换机，指定了Key为error
13.         channel.queueBind(MQUtils.QUEUE_NAME,MQUtils.EXCHANGE_NAME2,"error");
14.         //绑定队列2到交换机，指定了Key为debug
15.         channel.queueBind(MQUtils.QUEUE_NAME2,MQUtils.EXCHANGE_NAME2,"debug");
16.         //error是一个指定的Key
17.         channel.basicPublish(MQUtils.EXCHANGE_NAME2,"error",null,msg.getBytes());
18.         System.out.println("send:" + msg);
19.         channel.close();
20.         connection.close();
21.     }
22. }

复制代码

消费者1

1. /**
2. * 路由模式的消费者1
3. * 可以指定Key，消费特定的消息
4. */
5. public class RouteConsumer1 {
7.     public static void main(String[] args) throws IOException, InterruptedException {
8.         Connection connection = MQUtils.getConnection();
9.         Channel channel = connection.createChannel();
10.         channel.queueDeclare(MQUtils.QUEUE_NAME,false,false,false,null);
11.         QueueingConsumer queueingConsumer = new QueueingConsumer(channel);
12.         channel.basicConsume(MQUtils.QUEUE_NAME,true,queueingConsumer);
13.         while(true){
14.             QueueingConsumer.Delivery delivery = queueingConsumer.nextDelivery();
15.             System.out.println("RouteConsumer1 receive :" + new String(delivery.getBody()));
16.         }
17.     }
18. }

复制代码

消费者2

1. /**
2. * 路由模式的消费者2
3. * 可以指定Key，消费特定的消息
4. */
5. public class RouteConsumer2 {
7.     public static void main(String[] args) throws IOException, InterruptedException {
8.         Connection connection = MQUtils.getConnection();
9.         Channel channel = connection.createChannel();
10.         channel.queueDeclare(MQUtils.QUEUE_NAME2,false,false,false,null);
11.         QueueingConsumer queueingConsumer = new QueueingConsumer(channel);
12.         channel.basicConsume(MQUtils.QUEUE_NAME2,true,queueingConsumer);
13.         while(true){
14.             QueueingConsumer.Delivery delivery = queueingConsumer.nextDelivery();
15.             System.out.println("RouteConsumer2 receive :" + new String(delivery.getBody()));
17.         }
18.     }
19. }

复制代码

主题模子

主题模式和路由模式差不多，在key中可以参加通配符：

• * 匹配任意一个单词 com.* ----> com.hopu com.blb com.baidu
  
• # 匹配.号隔开的0个或多个单词 com.# ---> com.hopu.net com.hopu com.163.xxx.xxx.xxx
  

案例代码

生产者代码

1. /**
2.   主题模式的生产者
3. */
4. public class TopicProductor {
6.     public static void main(String[] args) throws IOException, InterruptedException {
7.         Connection connection = MQUtils.getConnection();
8.         Channel channel = connection.createChannel();
9.         //声明交换机类型为topic
10.         channel.exchangeDeclare(MQUtils.EXCHANGE_NAME3,"topic");
11.         //绑定队列到交换机，最后指定了Key
12.         channel.queueBind(MQUtils.QUEUE_NAME,MQUtils.EXCHANGE_NAME3,"xray.#");
13.         //绑定队列到交换机，最后指定了Key
14.         channel.queueBind(MQUtils.QUEUE_NAME2,MQUtils.EXCHANGE_NAME3,"*.*.cn");
15.         String msg = "Hello-->Topic";
16.         channel.basicPublish(MQUtils.EXCHANGE_NAME3,"rabbit.com.cn",null,msg.getBytes());
17.         System.out.println("send:" + msg);
18.         channel.close();
19.         connection.close();
20.     }
21. }

复制代码

消费者1

1. /**
2. * 主题模式的消费者1 ，类似路由模式，可以使用通配符对Key进行筛选
3. *   #匹配1个或多个单词，*匹配一个单词
4. */
5. public class TopicConsumer1 {
7.     public static void main(String[] args) throws IOException, InterruptedException {
8.         Connection connection = MQUtils.getConnection();
9.         Channel channel = connection.createChannel();
10.         channel.queueDeclare(MQUtils.QUEUE_NAME,false,false,false,null);
11.         QueueingConsumer queueingConsumer = new QueueingConsumer(channel);
12.         channel.basicConsume(MQUtils.QUEUE_NAME,true,queueingConsumer);
13.         while(true){
14.             QueueingConsumer.Delivery delivery = queueingConsumer.nextDelivery();
15.             System.out.println("TopicConsumer1 receive :" + new String(delivery.getBody()));
16.         }
17.     }
18. }

复制代码

消费者2

1. /**
2. * 主题模式的消费者2
3. */
4. public class TopicConsumer2 {
6.     public static void main(String[] args) throws IOException, InterruptedException {
7.         Connection connection = MQUtils.getConnection();
8.         Channel channel = connection.createChannel();
9.         channel.queueDeclare(MQUtils.QUEUE_NAME2,false,false,false,null);
10.         QueueingConsumer queueingConsumer = new QueueingConsumer(channel);
11.         channel.basicConsume(MQUtils.QUEUE_NAME2,true,queueingConsumer);
12.         while(true){
13.             QueueingConsumer.Delivery delivery = queueingConsumer.nextDelivery();
14.             System.out.println("TopicConsumer2 receive :" + new String(delivery.getBody()));
15.         }
16.     }
17. }

复制代码

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