RabbitMQ的5种模式——再探RabbitMQ的模式，简单、工作，发布订阅（广播），路由、主题 & 页面分析

叁歲伎倆 2024-02-21 07:27 15阅读 0赞

在这里插入图片描述

前言

RabbitMQ作为一款常用的消息中间件，在微服务项目中得到大量应用，其本身是微服务中的重点和难点，有不少概念我自己的也是一知半解，本系列博客尝试结合实际应用场景阐述RabbitMQ的应用，分析其为什么使用，并给出怎么用的案例。

本篇博客结合场景来阐述RabbitMQ的几种模式，描述了不同模式的应用场景，并给出相应的代码。（文末有惊喜~）

在这里插入图片描述
其他相关的rabbitmq博客文章列表如下：

RabbitMQ基础（1）——生产者消费者模型 & RabbitMQ简介 & Docker版本的安装配置 & RabbitMQ的helloworld + 分模块构建 & 解决大量注册案例

RabbitMQ基础（2）——发布订阅/fanout模式 & topic模式 & rabbitmq回调确认 & 延迟队列（死信）设计

RabbitMQ的Docker版本安装 + 延迟插件安装 & QQ邮箱和阿里云短信验证码的主题模式发送

目录

  • 前言
  • 引出
  • MQ场景：
    • 1异步处理
    • 2解耦
    • 3削峰填谷
    • 常见的MQ
  • 拆解控制台页面
    • 总览页面
    • 连接connection
    • 队列页面
  • 简单模式
    • 工具类：建立连接
    • 生产者：生产消息
    • 消费者：消费消息
    • 进行测试
      • Ack：项目中必须false
  • 工作模式
    • 一个生产者，两个消费者
    • QOS: 限流
      • Tips：队列参数怎么变？
    • 队列相关参数
      • x-max-length
      • x-overflow
      • x-max-length-bytes
  • 发布者订阅模式
    • 生产者：给fanout交换机发消息
    • 消费者1：队列q32
    • 消费者2：队列q321
  • 路由模式
    • 生产者：发给交换机，告诉路由键
    • 消费者1：根据路由键，接收3种消息
    • 消费者2：根据路由键，接收1种消息
  • 主题模式(Topic)
    • 生产者：主题交换机，带路由键
    • 消费者1：通配符的路由键
    • 消费者2：统配符
  • 总结

引出

1.MQ，消息队列的应用场景，几种MQ简单对比；
2.分析RabbitMQ的浏览器控制台页面；
3.结合场景来阐述RabbitMQ的几种模式，描述了不同模式的应用场景，并给出相应的代码；

在这里插入图片描述

MQ场景：

Message Queue 消息队列。

有比较大的负载，而且这些负载不用立刻马上给程序返回结构，可以有一些等待时间。可以用消息队列。

1异步处理

过去的项目，大部分都是同步解决问题。

在这里插入图片描述

  1. UserinfoService{
  3.     register(){
  5.         //典型的同步
  6.         //插入数据库
  7.         //发短信
  8.         //发邮件
  9.     }
  10. }
  11. //如果你的操作基本没有任何延时操作，或者瓶颈，没有压力。那么你没有必要用MQ
  12. UserinfoService{
  14.     register(){
  16.         //发起的是一个异步请求
  17.         都不等待对方给我返回的结果
  18.         //异步插入数据库
  19.         //异步发短信
  20.         //异步发邮件
  21.     }
  22. }

2解耦

在这里插入图片描述

库存有没有可能挂了。或者访问量巨大。因为库存慢，导致订单也慢。

在这里插入图片描述

解耦。

在这里插入图片描述

3削峰填谷

双12的时候，叫好早上的8点。秒杀的时候。

有些瞬间服务器压力是超大的，过了这个瞬间，几乎没有消耗量。

等服务器能够正常的时候，我慢慢执行就行了。

在这里插入图片描述

常见的MQ

MQ的前身，就是一个发布者订阅模式。

Kafka RabbitMQ（1W/s） RocketMQ ActiveMQ

Kafka : 10w/S 主要用于日志。

Kafka是由Apache软件基金会开发的一个开源流处理平台，由Scala和Java编写。Kafka是一种高吞吐量的分布式发布订阅消息系统，它可以处理消费者在网站中的所有动作流数据。 这种动作（网页浏览，搜索和其他用户的行动）是在现代网络上的许多社会功能的一个关键因素。 这些数据通常是由于吞吐量的要求而通过处理日志和日志聚合来解决。 对于像Hadoop一样的日志数据和离线分析系统，但又要求实时处理的限制，这是一个可行的解决方案。Kafka的目的是通过Hadoop的并行加载机制来统一线上和离线的消息处理，也是为了通过集群来提供实时的消息。

在这里插入图片描述

拆解控制台页面

总览页面

包括刷新时间间隔，rabbitmq节点，连接端口的信息等

在这里插入图片描述

配置可以导出和导入

在这里插入图片描述

连接connection

在这里插入图片描述

队列页面

包括队列的名字，状态，准备好的消息数量，未确认的消息数量，总计消息数量

在这里插入图片描述

简单模式

在这里插入图片描述

使用的依赖

  1. <!-- https://mvnrepository.com/artifact/com.rabbitmq/amqp-client -->
  2.     <dependency>
  3.         <groupId>com.rabbitmq</groupId>
  4.         <artifactId>amqp-client</artifactId>
  5.         <version>5.7.3</version>
  6.     </dependency>

工具类：建立连接

在这里插入图片描述

  1. package com.tianju.config;
  4. import com.rabbitmq.client.Connection;
  6. import java.io.IOException;
  7. import java.util.concurrent.TimeoutException;
  9. /**
  10.  * 建立连接的工具类
  11.  */
  12. public class ConnectionFactory {
  15.     public static Connection createConnection() throws IOException, TimeoutException {
  17.         com.rabbitmq.client.ConnectionFactory connectionFactory = new com.rabbitmq.client.ConnectionFactory();
  18.         connectionFactory.setHost("192.168.111.130"); // http://192.168.111.130/
  19.         connectionFactory.setPort(5672);
  20.         connectionFactory.setUsername("admin");
  21.         connectionFactory.setPassword("123");
  22.         connectionFactory.setVirtualHost("/demo");
  24.         // amqp://admin@192.168.111.130:5672/
  25.         return connectionFactory.newConnection();
  26.     }
  27. }

在这里插入图片描述

生产者：生产消息

在这里插入图片描述

  1. package com.tianju.simple;
  3. import com.rabbitmq.client.Channel;
  4. import com.rabbitmq.client.Connection;
  5. import com.tianju.config.ConnectionFactory;
  6. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  8. import java.io.IOException;
  9. import java.util.concurrent.TimeoutException;
  11. /**
  12.  * 生产者发送消息
  13.  * 建立连接--> 创建频道 --> 创建队列 --> 发送消息
  14.  */
  15. public class Provider {
  17.     private static String QUEUE_ORDER = "queue_order";
  18.     public static void main(String[] args) throws IOException, TimeoutException {
  20.         // 建立连接
  21.         Connection connection = ConnectionFactory.createConnection();
  22.         // 创建频道
  23.         Channel channel = connection.createChannel();
  24.         // 创建队列
  25.         channel.queueDeclare(QUEUE_ORDER,false,false,false,null);
  27.         // 发送消息，指定给哪个队列上发消息
  28.         for (int i = 0; i < 100; i++) {
  30.             String msg = "hello rabbitmq--"+i;
  31.             channel.basicPublish("", QUEUE_ORDER, null, msg.getBytes());
  33.             System.out.println("消息发布成功");
  35.         }
  36.         connection.close();
  37.     }
  38. }

消费者：消费消息

在这里插入图片描述

  1. package com.tianju.simple;
  3. import com.rabbitmq.client.*;
  4. import com.tianju.config.ConnectionFactory;
  6. import java.io.IOException;
  7. import java.util.concurrent.TimeoutException;
  9. public class Consumer {
  11.     private static String QUEUE_ORDER = "queue_order";
  12.     public static void main(String[] args) throws IOException, TimeoutException {
  14.         // 建立连接
  15.         Connection connection = ConnectionFactory.createConnection();
  16.         // 创建频道
  17.         Channel channel = connection.createChannel();
  18.         // 创建队列
  19.         channel.queueDeclare(QUEUE_ORDER,false,false,false,null);
  20.         // 队列必须声明，如果不存在，则自动创建
  22.         // 声明一个消费者
  23.         DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
  25.             /**
  26.              * 回调函数，用来接收消息
  27.              * @param consumerTag the <i>consumer tag</i> associated with the consumer
  28.              * @param envelope packaging data for the message
  29.              * @param properties content header data for the message
  30.              * @param body the message body (opaque, client-specific byte array)，字节数组
  31.              * @throws IOException
  32.              */
  33.             public void handleDelivery(String consumerTag,
  34.                                        Envelope envelope,
  35.                                        AMQP.BasicProperties properties,
  36.                                        byte[] body)
  37.                     throws IOException
  38.             {
  40.                 // no work to do
  42.                 String msg = new String(body); // 收到的信息
  43.                 System.out.println("消费者接收到："+msg);
  44.                 try {
  46.                     Thread.sleep(3000);// 模拟一个耗时操作
  47.                 } catch (InterruptedException e) {
  49.                     throw new RuntimeException(e);
  50.                 }
  51.             }
  52.         };
  53.         // 表明自己是消费者，接收消息
  54.         /**
  55.          * autoAck：自动确认设置成 true
  56.          */
  57.         channel.basicConsume(QUEUE_ORDER, true, defaultConsumer);
  58.     }
  59. }

进行测试

生产者发送100条消息
在这里插入图片描述

消费者进行消费，假设突然之间服务宕机了，此时消费了4条消息，理论上还应该有96条消息

在这里插入图片描述

打开控制台页面查看，消息全部消失了，出现了数据丢失的情况
！

在这里插入图片描述

全部用默认值的情况下，如果发生异常，则消息全部丢失。

消费者一次性拿到了所有的消息。

Ack：项目中必须false

生产者 ==投递消息=》队列

消费者=接受==》队列

ack: false 必须要手工确认。

消费者接到这个消息的时候， 这个消息进入 unack状态。

1：手工确认。消息删除。完成

2: 没手工确认，断开连接或者超时。

3：这个消息重新进入ready状态。等待其他消费者进行消费。

在这里插入图片描述

先设置自动ack为false，表示需要手工确认，然后在消费消息的方法中，进行消息的确认。

在这里插入图片描述

  1. package com.tianju.simple;
  3. import com.rabbitmq.client.*;
  4. import com.tianju.config.ConnectionFactory;
  6. import java.io.IOException;
  7. import java.util.concurrent.TimeoutException;
  9. public class Consumer {
  11.     private static String QUEUE_ORDER = "queue_order";
  12.     public static void main(String[] args) throws IOException, TimeoutException {
  14.         // 建立连接
  15.         Connection connection = ConnectionFactory.createConnection();
  16.         // 创建频道
  17.         Channel channel = connection.createChannel();
  18.         // 创建队列
  19.         channel.queueDeclare(QUEUE_ORDER,false,false,false,null);
  20.         // 队列必须声明，如果不存在，则自动创建
  22.         // 声明一个消费者
  23.         DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
  25.             /**
  26.              * 回调函数，用来接收消息
  27.              * @param consumerTag the <i>consumer tag</i> associated with the consumer
  28.              * @param envelope packaging data for the message
  29.              * @param properties content header data for the message
  30.              * @param body the message body (opaque, client-specific byte array)，字节数组
  31.              * @throws IOException
  32.              */
  33.             public void handleDelivery(String consumerTag,
  34.                                        Envelope envelope,
  35.                                        AMQP.BasicProperties properties,
  36.                                        byte[] body)
  37.                     throws IOException {
  39.                 // no work to do
  41.                 String msg = new String(body); // 收到的信息
  42.                 System.out.println("消费者接收到："+msg);
  43.                 try {
  45.                     Thread.sleep(3000);// 模拟一个耗时操作
  46.                 } catch (InterruptedException e) {
  48.                     throw new RuntimeException(e);
  49.                 }
  50.                 // 消费者代码可能有失败，消息拿到之后，可能还没有处理，就宕机了
  51.                 // 消息确认的代码一定在最后一行
  52.                 // long deliveryTag 消息的下标；
  53.                 // boolean multiple 是否批量确认；
  54.                 channel.basicAck(envelope.getDeliveryTag(),true); // 确认消息，批量确认
  55.             }
  56.         };
  57.         // 表明自己是消费者，接收消息
  58.         /**
  59.          * autoAck：自动确认设置成 true
  60.          * 是否自动确认，
  61.          * false：不进行自动确认；true：自动确认
  62.          * 消费过程中可能产生异常
  63.          * 如果产生异常，则必须进行消费补偿
  64.          */
  65.         channel.basicConsume(QUEUE_ORDER, false, defaultConsumer); // 接收消息
  66.     }
  67. }

再次模拟宕机的情况，一开始消息全部被放到Unack中，当宕机时，又把消息吐了出来，至少消息没有出现丢失的情况。

在这里插入图片描述

工作模式

在这里插入图片描述

一个生产者，两个消费者

在这里插入图片描述

两个消费者消费的是同一个队列中的消息。

两个消费者上来后，默认是按照平均分配。结果： 有人瞬间干完。有人很久都没干完。

QOS: 限流

可以限制你一次拉取几个。

这样两个消费者，也能够节省服务器CPU了。

  1. // 创建频道
  2.         Channel channel = connection.createChannel();
  3.         channel.basicQos(1); /** 1次只能拿1个 **/

在这里插入图片描述

当消费的速度太慢，觉得不够，加入其他的消费者。

核心，只有一个消息队列。

消息是轮询的方式，发送给不同的消费者。

在这里插入图片描述

Tips：队列参数怎么变？

队列的参数发生变化后，要删除再添加。

在这里插入图片描述

  1. // 创建频道
  2.         Channel channel = connection.createChannel();
  4.         channel.queueDelete(QUEUE_ORDER); // 先删除旧的队列
  6.         boolean durable = true; // 当前队列中的消息持久化操作，重启之后，消息还在
  7.         // 创建队列
  8.         channel.queueDeclare(QUEUE_ORDER,durable,false,false,null);

在这里插入图片描述

队列相关参数

x-max-length

在这里插入图片描述

  1. package com.tianju.work;
  3. import com.rabbitmq.client.Channel;
  4. import com.rabbitmq.client.Connection;
  5. import com.tianju.config.ConnectionFactory;
  7. import java.io.IOException;
  8. import java.util.HashMap;
  9. import java.util.Map;
  10. import java.util.concurrent.TimeoutException;
  12. /**
  13.  * 生产者发送消息
  14.  * 建立连接--> 创建频道 --> 创建队列 --> 发送消息
  15.  */
  16. public class Provider {
  18.     private static String QUEUE_ORDER = "queue_order";
  19.     public static void main(String[] args) throws IOException, TimeoutException {
  21.         // 建立连接
  22.         Connection connection = ConnectionFactory.createConnection();
  23.         // 创建频道
  24.         Channel channel = connection.createChannel();
  25.         channel.queueDelete(QUEUE_ORDER); // 先删除旧的队列
  26.         boolean durable = true; // 当前队列中的消息持久化操作，重启之后，消息还在
  28.         Map map = new HashMap();
  29.         map.put("x-max-length", 10); // 设置最大的长度，只接受10个
  30.         // 创建队列
  31.         channel.queueDeclare(QUEUE_ORDER,durable,false,false,map);
  33.         // 发送消息，指定给哪个队列上发消息
  34.         for (int i = 0; i < 1000; i++) {
  36.             String msg = "hello rabbitmq--"+i;
  37.             channel.basicPublish("", QUEUE_ORDER, null, msg.getBytes());
  39.             System.out.println("消息发布成功");
  40.         }
  41.         connection.close();
  42.     }
  43. }

How many (ready) messages a queue can contain before it starts to drop them from its head.

在这里插入图片描述

x-overflow

Sets the queue overflow behaviour. This determines what happens to messages when the maximum length of a queue is reached. Valid values are drop-head, reject-publish or reject-publish-dlx. The quorum queue type only supports drop-head and reject-publish.

在这里插入图片描述
此时就是前10个了

在这里插入图片描述

x-max-length-bytes

在这里插入图片描述

  1. package com.tianju.work;
  3. import com.rabbitmq.client.Channel;
  4. import com.rabbitmq.client.Connection;
  5. import com.tianju.config.ConnectionFactory;
  7. import java.io.IOException;
  8. import java.util.HashMap;
  9. import java.util.Map;
  10. import java.util.concurrent.TimeoutException;
  12. /**
  13.  * 生产者发送消息
  14.  * 建立连接--> 创建频道 --> 创建队列 --> 发送消息
  15.  */
  16. public class Provider {
  18.     private static String QUEUE_ORDER = "queue_order";
  19.     public static void main(String[] args) throws IOException, TimeoutException {
  21.         // 建立连接
  22.         Connection connection = ConnectionFactory.createConnection();
  23.         // 创建频道
  24.         Channel channel = connection.createChannel();
  25.         channel.queueDelete(QUEUE_ORDER); // 先删除旧的队列
  26.         boolean durable = true; // 当前队列中的消息持久化操作，重启之后，消息还在
  28.         Map map = new HashMap();
  29.         map.put("x-max-length", 10); // 设置最大的长度，只接受10个
  30.         map.put("x-overflow", "reject-publish"); // 拒绝发布，变成前10ge
  31.         map.put("x-max-length-bytes", 4); // 消息的最大字节长度
  32.         // 创建队列
  33.         channel.queueDeclare(QUEUE_ORDER,durable,false,false,map);
  35.         // 发送消息，指定给哪个队列上发消息
  36.         for (int i = 0; i < 20; i++) {
  38.             String msg = "53"+i;
  39.             channel.basicPublish("", QUEUE_ORDER, null, msg.getBytes());
  40.             msg = "hello rabbitmq--"+i;
  41.             channel.basicPublish("", QUEUE_ORDER, null, msg.getBytes());
  43.             System.out.println("消息发布成功");
  44.         }
  45.         connection.close();
  46.     }
  47. }

在这里插入图片描述

参数

在这里插入图片描述

  1. //创建队列
  2.         boolean durable = true;//当前队列中的消息进行持久化操作  重启之后，消息还在
  3.         Map map = new HashMap();
  4.         map.put("x-max-length",10);//设置队列的最大长度
  5.         map.put("x-overflow","reject-publish");//设置队列的最大长度  drop head
  6.         map.put("x-max-length-bytes",4);//设置消息的最大字节长度
  7. //        map.put("x-expires",10000);//超时后，直接删除队列
  8.         channel.queueDeclare(QUQUENAME,durable,false,false,map);

发布者订阅模式

在这里插入图片描述

工作模式中，只有一个队列。两个消费者轮询从同一个队列中取数据。

发布者订阅模式： 交换机后面会绑定多个消息队列。每个消息队列都有完整的信息。每个队列后的消费者都会有完整的消息。

交换机：就是一个无意识的广播。行为就是扇出.

在这里插入图片描述

场景：

下订单：

1：订单的数据入数据库
2：发送订单的短信
3： 物流

要每个队列都有完整的信息。

在这里插入图片描述

生产者：给fanout交换机发消息

  1. package com.tianju.publish;
  3. import com.rabbitmq.client.Channel;
  4. import com.rabbitmq.client.Connection;
  5. import com.rabbitmq.client.MessageProperties;
  6. import com.tianju.config.ConnectionFactory;
  8. import java.io.IOException;
  9. import java.util.concurrent.TimeoutException;
  11. /**
  12.  * 生产者发送消息
  13.  * 建立连接--> 创建频道 --> 创建队列 --> 发送消息
  14.  */
  15. public class Provider {
  17.     private static String QUEUE_ORDER = "queue_order";
  18.     private static  String EXCHANGE = "pet_exchange";
  19.     public static void main(String[] args) throws IOException, TimeoutException {
  21.         // 建立连接
  22.         Connection connection = ConnectionFactory.createConnection();
  23.         // 创建频道
  24.         Channel channel = connection.createChannel();
  26.         // 创建交换机
  27.         channel.exchangeDeclare(EXCHANGE, "fanout"); // 扇出，类型只能用这个
  29.         for (int i = 0; i < 100; i++) {
  31.             channel.basicPublish(EXCHANGE, "", MessageProperties.TEXT_PLAIN, ("hello fanout"+i).getBytes());
  32.         }
  33.     }
  34. }

消费者1：队列q32

  1. package com.tianju.publish;
  3. import com.rabbitmq.client.*;
  4. import com.tianju.config.ConnectionFactory;
  6. import java.io.IOException;
  7. import java.util.concurrent.TimeoutException;
  9. public class Consumer {
  11.     private static  String EXCHANGE = "pet_exchange";
  13.     public static void main(String[] args) throws IOException, TimeoutException {
  15.         // 建立连接
  16.         Connection connection = ConnectionFactory.createConnection();
  17.         // 创建频道
  18.         Channel channel = connection.createChannel();
  20.         // 创建交换机
  21.         channel.exchangeDeclare(EXCHANGE, "fanout"); // 扇出，类型只能用这个
  23.         channel.queueDeclare("q32", false, false, false, null);
  24.         channel.queueBind("q32", EXCHANGE, ""); // 路由键fanout模式，必须为空，即使写了是无效的
  26.         channel.basicConsume("q32", new DefaultConsumer(channel){
  28.             public void handleDelivery(String consumerTag,
  29.                                        Envelope envelope,
  30.                                        AMQP.BasicProperties properties,
  31.                                        byte[] body)
  32.                     throws IOException
  33.             {
  35.                 // no work to do
  36.                 String msg = new String(body);
  37.                 System.out.println("消费者1："+msg);
  38.                 channel.basicAck(envelope.getDeliveryTag(), false);
  39.             }
  40.         });
  41.     }
  42. }

消费者2：队列q321

  1. package com.tianju.publish;
  3. import com.rabbitmq.client.*;
  4. import com.tianju.config.ConnectionFactory;
  6. import java.io.IOException;
  7. import java.util.concurrent.TimeoutException;
  9. public class Consumer2 {
  11.     private static  String EXCHANGE = "pet_exchange";
  13.     public static void main(String[] args) throws IOException, TimeoutException {
  15.         // 建立连接
  16.         Connection connection = ConnectionFactory.createConnection();
  17.         // 创建频道
  18.         Channel channel = connection.createChannel();
  20.         // 创建交换机
  21.         channel.exchangeDeclare(EXCHANGE, "fanout"); // 扇出，类型只能用这个
  23.         channel.queueDeclare("q321", false, false, false, null);
  24.         channel.queueBind("q321", EXCHANGE, ""); // 路由键fanout模式，必须为空，即使写了是无效的
  26.         channel.basicConsume("q321", new DefaultConsumer(channel){
  28.             public void handleDelivery(String consumerTag,
  29.                                        Envelope envelope,
  30.                                        AMQP.BasicProperties properties,
  31.                                        byte[] body)
  32.                     throws IOException
  33.             {
  35.                 // no work to do
  36.                 String msg = new String(body);
  37.                 System.out.println("消费者2："+msg);
  38.                 channel.basicAck(envelope.getDeliveryTag(), false);
  39.             }
  40.         });
  41.     }
  42. }

路由模式

允许用不同的路由键来接受不同的信息。

在这里插入图片描述

队列会接到完整的全部的信息。

日志系统：

日志级别：

OFF INFO DEBUG ERROR FATAL ALL

如果是致命错误： 立刻给运维发短信。

全部信息： 放到专门的日志系统
Error: 进行发邮件

在这里插入图片描述

生产者：发给交换机，告诉路由键

  1. package com.tianju.routing;
  3. import com.rabbitmq.client.Channel;
  4. import com.rabbitmq.client.Connection;
  5. import com.rabbitmq.client.MessageProperties;
  6. import com.tianju.config.ConnectionFactory;
  8. import java.io.IOException;
  9. import java.util.concurrent.TimeoutException;
  11. /**
  12.  * 生产者发送消息
  13.  * 建立连接--> 创建频道 --> 创建队列 --> 发送消息
  14.  */
  15. public class Provider {
  17.     private static String EXCHANGE = "exchange_58";
  18.     public static void main(String[] args) throws IOException, TimeoutException {
  20.         // 建立连接
  21.         Connection connection = ConnectionFactory.createConnection();
  22.         // 创建频道
  23.         Channel channel = connection.createChannel();
  25.         // 创建交换机
  26.         channel.exchangeDeclare(EXCHANGE, "direct"); // 直接交换机
  29.         String msg  = "this is fatal message";
  30.         channel.basicPublish(EXCHANGE,"fatal", MessageProperties.TEXT_PLAIN,msg.getBytes());
  31.         msg  = "this is error message";
  32.         channel.basicPublish(EXCHANGE,"error", MessageProperties.TEXT_PLAIN,msg.getBytes());
  33.         msg  = "this is debug message";
  34.         channel.basicPublish(EXCHANGE,"debug", MessageProperties.TEXT_PLAIN,msg.getBytes());
  36.     }
  37. }

消费者1：根据路由键，接收3种消息

  1. package com.tianju.routing;
  3. import com.rabbitmq.client.*;
  4. import com.tianju.config.ConnectionFactory;
  6. import java.io.IOException;
  7. import java.util.Calendar;
  8. import java.util.concurrent.TimeoutException;
  10. public class Consumer {
  12.     private static String EXCHANGE = "exchange_58";
  13.     public static void main(String[] args) throws IOException, TimeoutException {
  15.         // 建立连接
  16.         Connection connection = ConnectionFactory.createConnection();
  17.         // 创建频道
  18.         Channel channel = connection.createChannel();
  19.         // 创建队列
  20.         // 创建交换机
  21.         channel.exchangeDeclare(EXCHANGE, "direct"); // 直接交换机
  22.         // 队列必须声明，如果不存在，则自动创建
  24.         channel.queueDeclare("q58", false, false, false, null);
  26.         channel.queueBind("q58", EXCHANGE, "fatal");
  27.         channel.queueBind("q58", EXCHANGE, "error");
  28.         channel.queueBind("q58", EXCHANGE, "debug");
  30.         channel.basicConsume("q58", new DefaultConsumer(channel){
  32.             public void handleDelivery(String consumerTag,
  33.                                        Envelope envelope,
  34.                                        AMQP.BasicProperties properties,
  35.                                        byte[] body)
  36.                     throws IOException
  37.             {
  39.                 // no work to do
  40.                 String msg = new String(body);
  41.                 System.out.println("消费者1："+msg);
  42.                 channel.basicAck(envelope.getDeliveryTag(), false);
  43.             }
  44.         });
  46.     }
  47. }

消费者2：根据路由键，接收1种消息

  1. package com.tianju.routing;
  3. import com.rabbitmq.client.*;
  4. import com.tianju.config.ConnectionFactory;
  6. import java.io.IOException;
  7. import java.util.concurrent.TimeoutException;
  9. public class Consumer2 {
  11.     private static String EXCHANGE = "exchange_58";
  12.     public static void main(String[] args) throws IOException, TimeoutException {
  14.         // 建立连接
  15.         Connection connection = ConnectionFactory.createConnection();
  16.         // 创建频道
  17.         Channel channel = connection.createChannel();
  18.         // 创建队列
  19.         // 创建交换机
  20.         channel.exchangeDeclare(EXCHANGE, "direct"); // 直接交换机
  21.         // 队列必须声明，如果不存在，则自动创建
  23.         channel.queueDeclare("q581", false, false, false, null);
  25.         channel.queueBind("q581", EXCHANGE, "fatal");
  27.         channel.basicConsume("q581", new DefaultConsumer(channel){
  29.             public void handleDelivery(String consumerTag,
  30.                                        Envelope envelope,
  31.                                        AMQP.BasicProperties properties,
  32.                                        byte[] body)
  33.                     throws IOException
  34.             {
  36.                 // no work to do
  37.                 String msg = new String(body);
  38.                 System.out.println("消费者1："+msg);
  39.                 channel.basicAck(envelope.getDeliveryTag(), false);
  40.             }
  41.         });
  43.     }
  44. }

主题模式(Topic)

在这里插入图片描述

可以采取通配符模式：

  1. * (star) can substitute for exactly one word.
  2. # (hash) can substitute for zero or more words.

#`：匹配0个或多个词

*：匹配不多不少恰好1个词

举例：

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

item.*：只能匹配item.insert

场景：

京东： 无锡，河南下订单。退货

==========总部 收到这个订单 .

无锡分销商也有这个消息 *.wuxi

在这里插入图片描述

生产者：主题交换机，带路由键

  1. package com.tianju.topic;
  3. import com.rabbitmq.client.Channel;
  4. import com.rabbitmq.client.Connection;
  5. import com.rabbitmq.client.MessageProperties;
  6. import com.tianju.config.ConnectionFactory;
  8. import java.io.IOException;
  9. import java.util.concurrent.TimeoutException;
  11. /**
  12.  * 生产者发送消息
  13.  * 建立连接--> 创建频道 --> 创建队列 --> 发送消息
  14.  */
  15. public class Provider {
  17.     private static String EXCHANGE = "exchange_58";
  18.     public static void main(String[] args) throws IOException, TimeoutException {
  20.         // 建立连接
  21.         Connection connection = ConnectionFactory.createConnection();
  22.         // 创建频道
  23.         Channel channel = connection.createChannel();
  25.         // 创建交换机
  26.         channel.exchangeDeclare(EXCHANGE, "topic"); // 主题模式交换机
  29.         String msg  = "this is wuxi order";
  30.         channel.basicPublish(EXCHANGE,"order.wuxi", MessageProperties.TEXT_PLAIN,msg.getBytes());
  31.         msg  = "this is henan order";
  32.         channel.basicPublish(EXCHANGE,"order.henan", MessageProperties.TEXT_PLAIN,msg.getBytes());
  33.         msg  = "this is wuxi back";
  34.         channel.basicPublish(EXCHANGE,"back.wuxi", MessageProperties.TEXT_PLAIN,msg.getBytes());
  35.         msg  = "this is henan back";
  36.         channel.basicPublish(EXCHANGE,"back.henan", MessageProperties.TEXT_PLAIN,msg.getBytes());
  38.     }
  39. }

消费者1：通配符的路由键

  1. package com.tianju.topic;
  3. import com.rabbitmq.client.*;
  4. import com.tianju.config.ConnectionFactory;
  6. import java.io.IOException;
  7. import java.util.concurrent.TimeoutException;
  9. public class Consumer {
  11.     private static String EXCHANGE = "exchange_58";
  12.     public static void main(String[] args) throws IOException, TimeoutException {
  14.         // 建立连接
  15.         Connection connection = ConnectionFactory.createConnection();
  16.         // 创建频道
  17.         Channel channel = connection.createChannel();
  18.         // 创建队列
  19.         // 创建交换机
  20.         channel.exchangeDeclare(EXCHANGE, "topic"); // 主题交换机
  21.         // 队列必须声明，如果不存在，则自动创建
  23.         channel.queueDeclare("q58", false, false, false, null);
  25.         channel.queueBind("q58", EXCHANGE, "*.wuxi"); // 意味着无锡的订单和退货都能收到
  28.         channel.basicConsume("q58", new DefaultConsumer(channel){
  30.             public void handleDelivery(String consumerTag,
  31.                                        Envelope envelope,
  32.                                        AMQP.BasicProperties properties,
  33.                                        byte[] body)
  34.                     throws IOException
  35.             {
  37.                 // no work to do
  38.                 String msg = new String(body);
  39.                 System.out.println("无锡仓库："+msg);
  40.                 channel.basicAck(envelope.getDeliveryTag(), false);
  41.             }
  42.         });
  44.     }
  45. }

消费者2：统配符

  1. package com.tianju.topic;
  3. import com.rabbitmq.client.*;
  4. import com.tianju.config.ConnectionFactory;
  6. import java.io.IOException;
  7. import java.util.concurrent.TimeoutException;
  9. public class Consumer2 {
  11.     private static String EXCHANGE = "exchange_58";
  12.     public static void main(String[] args) throws IOException, TimeoutException {
  14.         // 建立连接
  15.         Connection connection = ConnectionFactory.createConnection();
  16.         // 创建频道
  17.         Channel channel = connection.createChannel();
  18.         // 创建队列
  19.         // 创建交换机
  20.         channel.exchangeDeclare(EXCHANGE, "topic"); // 主题交换机
  21.         // 队列必须声明，如果不存在，则自动创建
  23.         channel.queueDeclare("q581", false, false, false, null);
  25.         channel.queueBind("q581", EXCHANGE, "*.*"); // 意味着所有的订单和退货都能收到
  28.         channel.basicConsume("q581", new DefaultConsumer(channel){
  30.             public void handleDelivery(String consumerTag,
  31.                                        Envelope envelope,
  32.                                        AMQP.BasicProperties properties,
  33.                                        byte[] body)
  34.                     throws IOException
  35.             {
  37.                 // no work to do
  38.                 String msg = new String(body);
  39.                 System.out.println("总部仓库："+msg);
  40.                 channel.basicAck(envelope.getDeliveryTag(), false);
  41.             }
  42.         });
  44.     }
  45. }

总结

1.MQ，消息队列的应用场景，几种MQ简单对比；
2.分析RabbitMQ的浏览器控制台页面；
3.结合场景来阐述RabbitMQ的几种模式，描述了不同模式的应用场景，并给出相应的代码；

在这里插入图片描述

发表评论

表情：
评论列表 （有 0 条评论，15人围观）

还没有评论，来说两句吧...

相关阅读