RabbitMQ（Direct 订阅模型-路由模式）的利用

  
RabbitMQ（Direct 订阅模型-路由模式）

一，Direct 订阅模型-路由模式介绍（Routing）

​ 订阅模型-路由模式，此时生产者发送消息时需要指定 RoutingKey，即路由 Key，Exchange 接收到消息时转发到与 RoutingKey 相匹配的队列中。
​ direct的意思是直接的，direct范例的Exchange会将消息转发到指定Routing key的Queue上，Routing key的剖析规则为准确匹配。也就是只有当producer发送的消息的Routing key与某个Binding key相等时，消息才会被分发到对应的Queue上。
路由模式:

1、每个消耗者监听自己的队列，并且设置routingkey;
2、生产者将消息发给交换机，由交换机根据routingkey来转发消息到指定的队列;
应用场景:用户通知，当用户充值成功或转账完成系统通知用户，通知方式有短信、邮件多种方法，
二，利用

1.添加依赖

1. <!--引入rabbitMQ的依赖-->
2. <dependency>
3.     <groupId>org.springframework.boot</groupId>
4.     <artifactId>spring-boot-starter-amqp</artifactId>
5. </dependency>

复制代码

2.修改设置文件

1.   rabbitmq:
2.     host: 192.168.193.131
3.     publisher-confirm-type: correlated  #开启异步消息确认
4.     publisher-returns: true #开启publisher-return的机制
5.     template:
6.       mandatory: true  #消息路由失败，调用ReturnCallback
7.     listener:
8.       simple:  #消息监听器的一种简单配置方式
9.         retry:
10.           enabled: true # 开启消费者失败重试
11.           initial-interval: 1000 # 初识的失败等待时长为1秒
12.           multiplier: 1 # 失败的等待时长倍数，下次等待时长 = multiplier * last-interval
13.           max-attempts: 3 # 最大重试次数
14.           stateless: true # true无状态；false有状态。如果业务中包含事务，这里改为false
15.         acknowledge-mode: manual
16.       direct: #特定的消息监听器配置类型或者是指定一个直连型的交换机），用于消息的路由和分发
17.         retry:
18.           enabled: true # 开启消费者失败重试
19.           initial-interval: 1000 # 初识的失败等待时长为1秒
20.           multiplier: 1 # 失败的等待时长倍数，下次等待时长 = multiplier * last-interval
21.           max-attempts: 3 # 最大重试次数
22.           stateless: true # true无状态；false有状态。如果业务中包含事务，这里改为false
23.         acknowledge-mode: manual #手动确认接收到的消息

复制代码

3.创建立置类

1. import org.springframework.amqp.core.Binding;
2. import org.springframework.amqp.core.BindingBuilder;
3. import org.springframework.amqp.core.DirectExchange;
4. import org.springframework.amqp.core.Queue;
5. import org.springframework.context.annotation.Bean;
6. import org.springframework.context.annotation.Configuration;
8. /**
9. * @Description:rabbitMQ配置类 会在应用启动时被加载。
10. * @bean  :用于指示一个方法将会产生一个 bean 对象，并添加到 Spring 容器中。
11. */
12. @Configuration
13. public class MyRabbitMQConfig {
16.     /**
17.      * @Description:创建一个名为 "loginQueue" 的 RabbitMQ 队列。
18.      * 参数 true 表示这个队列是持久化的，这意味着在 RabbitMQ 服务器重启后，这个队列仍然会存在，以保证消息不丢失
19.      */
20.     @Bean
21.     public Queue loginQueue(){
22.         /*第一个参数为队列的名称 第二个参数为true 表示队列持久化 保障消息不丢失*/
23.         return new Queue("loginQueue",true);
24.     }
26.     /**
27.      * @Description:方法用于创建一个名为 "directExchange" 的 Direct Exchange（直连交换机）。
28.      * Direct Exchange 是 RabbitMQ 中一种简单的交换机类型，它会根据消息的 routing key 将消息路由到相应的队列
29.      */
30.     @Bean
31.     public DirectExchange directExchange(){
32.         return new DirectExchange("directExchange");
33.     }
36.     /**
37.      * @Description:创建一个绑定，将之前创建的 "loginQueue" 队列绑定到 "directExchange" 交换机上，并且指定 routing key 为 "login"     
38.      */
39.     @Bean
40.     public Binding loginQueueBinding(){
41.         return BindingBuilder.bind(loginQueue()).to(directExchange()).with("login");
42.     }
44. }

复制代码

4.注入RabbitMQ模版引擎

1. //RabbitTemplate是Spring AMQP项目中的一个关键组件，用于在应用程序和RabbitMQ消息代理之间发送和接收消息。
2. @Autowired
3. private RabbitTemplate rabbitTemplate;

复制代码

5.消息的发送

1. //记录登陆日志 记录到RabbitMQ
2. //定义消息的唯一ID 防止消息重复消费
3. String msgId = "msg-" + UUID.randomUUID().toString();
4. //定义消息内容
5. String msgBody = JSON.toJSONString(tbUser);
7. //将消息唯一表示存入redis缓存  防止消息重复消费
8. stringRedisTemplate.opsForValue().set(msgId, msgBody);
10. /*组装消息体 发送消息队列*/
11. MessageVO messageVO = new MessageVO();
12. messageVO.setMsgID(msgId);
13. messageVO.setMsgBody(msgBody);
15. /*convertAndSend: 这是RabbitTemplate提供的一个方法，用于将消息转换并发送到RabbitMQ队列或交换机。
16.         "directExchange": 这是消息要发送到的目标交换机的名称。在RabbitMQ中，交换机是消息的分发中心，它根据特定的路由规则将消息发送到一个或多个队列。
17.         "login": 这是消息的路由键  routing key 。在直接交换机模式下，消息的路由键与队列的绑定键一致时，消息会被发送到该队列。
18.         JSON.toJSONString(messageVO): 这是要发送的消息内容。在这里，messageVO对象被转换成JSON格式的字符串，然后作为消息发送到RabbitMQ。
19.         这行代码的作用是将一个消息发送到名为directExchange的交换机，并且路由键为login，消息内容为messageVO对象的JSON字符串表示。*/
20. rabbitTemplate.convertAndSend("directExchange", "login", JSON.toJSONString(messageVO));

复制代码

6.消息的接收(监听)

1. /**
2.      * @param (org.springframework.amqp.core.Message)message ：RabbitMQ 消息的对象，包含消息体和消息属性等信息。
3.      * @param（com.rabbitmq.client.Channel;）channel：RabbitMQ 的通道，用于手动确认消息消费。
4.      * @return void
5.      * @date 2024/5/24 19:59
6.      * @Description: TODO
7.      * @RabbitListener 注解:主要用于消费名为 "loginQueue" 的队列中的消息。
8.      *
9.      * @RabbitListener(queues = "loginQueue")：这个注解用于声明一个 RabbitMQ 消息监听器，
10.      * 它会监听名为 "loginQueue" 的队列，一旦有消息到达这个队列，就会触发 recvLogMessage 方法进行消费。
11.      */
12.     @RabbitListener(queues = "loginQueue")
13.     public void recvLogMessage(Message message, Channel channel) {
14.         try {
15.             //-- 接收消息  获取消息体，然后使用 JSON 解析成 MessageVo 对象
16.             String s = new String(message.getBody());
17.             MessageVO messageVO = JSON.parseObject(s, MessageVO.class);
18.             System.out.println("RabbitMQ 收到消息：" + messageVO.getMsgID());
20.             //-- 判断消息重复消费
21.             String msgId = messageVO.getMsgID();
22.             if (!stringRedisTemplate.hasKey(msgId)) {
23.                 // 消息已经消费过了，删除消息
24.                 System.out.println("RabbitMQ 消息重复了：" + messageVO.getMsgID());
25.                 // channel: 这是一个通道对象，用于与消息队列服务进行通信。通常在消息队列框架中，你需要首先建立一个通道（channel），然后通过这个通道发送、接收消息等操作。
26.                 // basicAck: 这是一个确认消息的方法。在消息队列中，消费者（consumer）收到消息后，需要向消息队列服务确认（acknowledge）已经成功处理了这条消息，以便消息队列服务可以将这条消息标记为已处理。basicAck就是用来进行这个确认操作的方法。
27.                 // message.getMessageProperties().getDeliveryTag(): 这部分代码用于获取消息的交付标签（delivery tag）。消息队列服务在向消费者传递消息时，会给每条消息分配一个唯一的标识符，即交付标签。消费者收到消息后，可以通过这个标签来确认消息。
28.                 // true: 这个参数表示是否批量确认。如果设置为true，表示确认当前标签之前的所有未确认消息；如果设置为false，则仅确认当前标签所代表的消息。
29.                 // 综合起来，这行代码的作用是：使用给定的通道对象（channel）对收到的消息进行确认，确认方式是通过交付标签（delivery tag）来指定。
30.                 channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
31.                 return;
32.             }                                                
33.                               
34.             //-- 消费消息
35.             TbUser tbUser = JSON.parseObject(messageVO.getMsgBody(), TbUser.class);
36.             TbLog tbLog = new TbLog();
37.             tbLog.setCreateTime(new Date());
38.             tbLog.setLogContent(messageVO.getMsgBody());
40.             tbLogMapper.insert(tbLog);
42.             //-- 删除消息
43.             channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
45.             //-- 删除消息的唯一ID，防止重复消费
46.             stringRedisTemplate.delete(msgId);
47.             System.out.println("RabbitMQ 登录日志消息，消费成功了~~~~~~~~~~~~~~~~~~~~~~");
48.         } catch (IOException e) {
49.             System.out.println("登录日志接收失败了~~~~~~~~~~~~~~~~~~~~~");
50.             e.printStackTrace();
51.         }
52.     }

复制代码

7.设置回调函数 和 确认回调

1. /**
2.      * @param
3.      * @return void
4.      * @date 2024/5/24 20:03
5.      * @Description: TODO 这是一个自定义方法名，用于初始化RabbitMQ相关设置。
6.      * @PostConstruct: 这是一个Spring注解，用于标记在bean初始化之后立即执行的方法。
7.      */
8.     @PostConstruct
9.     public void initRabbitMQ() {
10.         System.out.println("设置RabbitMQ消息回调，确保消息不丢失");
11.         //-- return 回调 : 消息发送失败了，会被调用
12.         /*当消息发送失败时，RabbitMQ会调用此回调函数。回调函数会重新发送消息*/
13.         rabbitTemplate.setReturnCallback(
14.                 new RabbitTemplate.ReturnCallback() {
15.                     /*returnedMessage 方法中，包含了以下四个参数：
16.                                           Message message：表示发送失败的消息对象，包含消息的内容和元数据。
17.                       int replyCode：表示返回的错误码，用于指示消息发送失败的原因。
18.                       String replyText：表示返回的错误信息，用于指示消息发送失败的具体描述。
19.                       String exchange 和 String routingKey：分别表示发送消息时使用的交换机和路由键，用于指示消息发送的目的地。*/
20.                     @Override
21.                     public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
22.                         // 消息发送失败了，再次发送
23.                         System.out.println("RabbitMQ 消息发送失败，再次发送");
24.                         rabbitTemplate.convertAndSend(exchange, routingKey, message);
25.                     }
26.                 }
27.         );
29.         //-- confirm 回调 : 消息发送无论成功还是失败，都会被回调
30.         //当消息发送成功或失败时，RabbitMQ都会调用此回调函数。回调函数会根据ack参数判断消息是否成功发送。
31.         rabbitTemplate.setConfirmCallback(
32.                 new RabbitTemplate.ConfirmCallback() {
33.                     /*CorrelationData correlationData：这个参数是用来关联生产者发送的消息和消息发送确认结果的数据对象。在发送消息时，可以通过 CorrelationData 设置一个唯一的标识符来关联消息，在确认时可以通过这个标识符来确定是哪条消息的确认结果。
34.                    boolean ack：这个参数表示消息是否成功发送到 RabbitMQ 服务器并得到了确认。当 ack 为 true 时，表示消息发送成功；当 ack 为 false 时，表示消息发送失败。
35.                    String cause：这个参数用来描述消息发送失败的原因。当 ack 为 false 时，cause 参数会包含失败的具体原因，可以用来进行日志记录或者错误处理。*/
36.                     @Override
37.                     public void confirm(CorrelationData correlationData, boolean ack, String cause) {
38.                         if (ack) {
39.                             System.out.println("RabbitMQ 消息发送成功了");
40.                         } else {
41.                             System.out.println("RabbitMQ 消息发送失败了");
42.                         }
43.                     }
44.                 }
45.         );
46.     }

复制代码

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