威海市建设局网站,mip织梦手机网站模板,wordpress图创,多少钱一个网站RabbitMQ中的路由模式#xff08;Direct模式#xff09;应该是在实际工作中运用的比较多的一种模式了#xff0c;这个模式和发布与订阅模式的区别在于路由模式需要有一个routingKey#xff0c;在配置上#xff0c;交换机类型需要注入DirectExchange类型的交换机bean对象。…RabbitMQ中的路由模式Direct模式应该是在实际工作中运用的比较多的一种模式了这个模式和发布与订阅模式的区别在于路由模式需要有一个routingKey在配置上交换机类型需要注入DirectExchange类型的交换机bean对象。在交换机和队列的绑定过程中绑定关系需要在绑定一个路由key。由于在实际的工作中不大可能会用自动确认的模式所以我们在整合路由模式的过程中依然采用发送消息双确认机制和消费端手动确认的机制来保证消息的准确送达与消息防丢失。
1. 添加配置
在配置文件中配置rabbitmq的相关账号信息开启消息发送回调机制配置文件其实和发布订阅模式是一样的。配置详情如下
server:port: 10001spring:application:name: springboot-rabbitmq-s1rabbitmq:host: 127.0.0.1port: 5672virtual-host: /username: adminpassword: admin# 发送者开启 return 确认机制publisher-returns: true# 发送者开启 confirm 确认机制publisher-confirm-type: correlated2. 创建配置类 创建配置类RabbitMQConfig用于声明交换机、队列建立队列和交换机的绑定关系注入RabbitTemplate的bean对象。配置类详情如下package com.study.rabbitmq.config;import cn.hutool.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;/*** Author alen* DATE 2022/6/7 23:50*/
Slf4j
Configuration
public class RabbitMQConfig {public static final String EXCHANGE_NAME direct-order-exchange;public static final String SMS_QUEUE sms-direct-queue;public static final String EMAIL_QUEUE email-direct-queue;public static final String WECHAT_QUEUE wechat-direct-queue;/*** 1.* 声明交换机* return*/Beanpublic DirectExchange directExchange() {/*** directExchange的参数说明:* 1. 交换机名称* 2. 是否持久化 true持久化交换机一直保留 false不持久化用完就删除* 3. 是否自动删除 false不自动删除 true自动删除*/return new DirectExchange(EXCHANGE_NAME, true, false);}/*** 2.* 声明队列* return*/Beanpublic Queue smsQueue() {/*** Queue构造函数参数说明* 1. 队列名* 2. 是否持久化 true持久化 false不持久化*/return new Queue(SMS_QUEUE, true);}Beanpublic Queue emailQueue() {return new Queue(EMAIL_QUEUE, true);}Beanpublic Queue wechatQueue() {return new Queue(WECHAT_QUEUE, true);}/*** 3.* 队列与交换机绑定*/Beanpublic Binding smsBinding() {return BindingBuilder.bind(smsQueue()).to(directExchange()).with(sms);}Beanpublic Binding emailBinding() {return BindingBuilder.bind(emailQueue()).to(directExchange()).with(email);}Beanpublic Binding wechatBinding() {return BindingBuilder.bind(wechatQueue()).to(directExchange()).with(wechat);}/*** 将自定义的RabbitTemplate对象注入bean容器** param connectionFactory* return*/Beanpublic RabbitTemplate createRabbitTemplate(ConnectionFactory connectionFactory) {RabbitTemplate rabbitTemplate new RabbitTemplate();rabbitTemplate.setConnectionFactory(connectionFactory);//设置开启消息推送结果回调rabbitTemplate.setMandatory(true);//设置ConfirmCallback回调rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {log.info(ConfirmCallback start );log.info(回调数据{}, correlationData);log.info(确认结果{}, ack);log.info(返回原因{}, cause);log.info(ConfirmCallback end );}});//设置ReturnCallback回调rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {Overridepublic void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {log.info(ReturnCallback start );log.info(发送消息{}, JSONUtil.toJsonStr(message));log.info(结果状态码{}, replyCode);log.info(结果状态信息{}, replyText);log.info(交换机{}, exchange);log.info(路由key{}, routingKey);log.info(ReturnCallback end );}});return rabbitTemplate;}
}3. 消费者配置 在消费者项目的配置文件中开启手动确认配置详情如下server:port: 10002spring:application:name: springboot-rabbitmq-s2rabbitmq:host: 127.0.0.1port: 5672virtual-host: /username: adminpassword: adminlistener:simple:# 表示消费者消费成功消息以后需要手工的进行签收ack确认默认为 autoacknowledge-mode: manual4. 创建消费者
分别创建三个消费者DirectEmailConsumer、DirectSmsConsumer、DirectWechatConsumer来监听对应的队列有消息后进行消费三个消费者大同小异分别如下
4.1 DirectEmailConsumer
package com.study.rabbitmq.service.direct;import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;import java.io.IOException;/*** Author alen* DATE 2022/6/10 22:54*/
Slf4j
Service
RabbitListener(queues {email-direct-queue}) //监听队列
public class DirectEmailConsumer {//标记消费者逻辑执行方法RabbitHandlerpublic void emailMessage(String msg, Channel channel, Message message) throws IOException {try {log.info(Email direct --接收到消息{}, msg);channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);} catch (Exception e) {if (message.getMessageProperties().getRedelivered()) {log.error(消息已重复处理失败,拒绝再次接收...);//basicReject: 拒绝消息与basicNack区别在于不能进行批量操作其他用法很相似 false表示消息不再重新进入队列channel.basicReject(message.getMessageProperties().getDeliveryTag(), false); // 拒绝消息} else {log.error(消息即将再次返回队列处理...);// basicNack:表示失败确认一般在消费消息业务异常时用到此方法可以将消息重新投递入队列channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);}}}
}4.2 DirectSmsConsumer
package com.study.rabbitmq.service.direct;import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;import java.io.IOException;/*** Author alen* DATE 2022/6/10 22:55*/
Slf4j
Service
RabbitListener(queues {sms-direct-queue}) //监听队列
public class DirectSmsConsumer {RabbitHandlerpublic void smsMessage(String msg, Channel channel, Message message) throws IOException {try {log.info(sms direct --接收到消息{}, msg);channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);} catch (Exception e) {if (message.getMessageProperties().getRedelivered()) {log.error(消息已重复处理失败,拒绝再次接收...);//basicReject: 拒绝消息与basicNack区别在于不能进行批量操作其他用法很相似 false表示消息不再重新进入队列channel.basicReject(message.getMessageProperties().getDeliveryTag(), false); // 拒绝消息} else {log.error(消息即将再次返回队列处理...);// basicNack:表示失败确认一般在消费消息业务异常时用到此方法可以将消息重新投递入队列channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);}}}
}4.3 DirectWechatConsumer
package com.study.rabbitmq.service.direct;import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;import java.io.IOException;/*** Author chaoxian.wu* DATE 2022/6/10 22:55*/
Slf4j
Service
RabbitListener(queues {wechat-direct-queue}) //监听队列
public class DirectWechatConsumer {RabbitHandlerpublic void wechatlMessage(String msg, Channel channel, Message message) throws IOException {try {log.info(wechat direct --接收到消息{}, msg);channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);} catch (Exception e) {if (message.getMessageProperties().getRedelivered()) {log.error(消息已重复处理失败,拒绝再次接收...);//basicReject: 拒绝消息与basicNack区别在于不能进行批量操作其他用法很相似 false表示消息不再重新进入队列channel.basicReject(message.getMessageProperties().getDeliveryTag(), false); // 拒绝消息} else {log.error(消息即将再次返回队列处理...);// basicNack:表示失败确认一般在消费消息业务异常时用到此方法可以将消息重新投递入队列channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);}}}
}以上就是全部的代码部分接下来我们在进入测试看看实际效果如何先发布一个routingKeysms的消息查看是不是只有对应的一个队列中接收到消息消息发送详情
package com.study.rabbitmq;import com.study.rabbitmq.entity.Order;
import com.study.rabbitmq.service.OrderService;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;import java.util.UUID;SpringBootTest
class SpringbootRabbitmqS1ApplicationTests {Autowiredprivate OrderService orderService;Testvoid contextLoads() {for (long i 1; i 2; i) {//交换机名称String exchangeName direct-order-exchange;//路由keyString routingKey sms;Order order buildOrder(i);orderService.createOrder(order, routingKey, exchangeName);}}private Order buildOrder(long id) {Order order new Order();order.setRequestId(id);order.setUserId(id);order.setOrderNo(UUID.randomUUID().toString());order.setAmount(10L);order.setGoodsNum(1);order.setTotalAmount(10L);return order;}
}我们登录rabbitmq管理后台查看下只有sms-direct-queue这个队列有一条消息效果如下 我们启动消费者看下是不是只有监听了sms-direct-queue这个队列的消费者有消费日志效果如下 再发一条routingKeyemail的消息消费的日志效果图示如下 到此其实已经springboot整合rabbitmq的路由模式结束了这种模式在工作中还是比较常见的我们演示的是单点的效果实际工作中不大可能会使用服务单点部署现在都讲究服务的高可用就得服务集群部署又会涉及到消息重复消费的问题需要处理我个人觉得遇到重复消费问题我第一时间想到的就是分布式锁哈哈~。但是锁什么呢肯定是消息中的具备唯一性的属性。来达到防止消息的重复消费。
整个过程中其实还存在一个小问题没有验证就是ReturnCallback回调机制没有触发因为这个得发生在交换机将消息发送到队列的时候失败才会触发那么我们就发送一个不存在的routingKey就可以触发了我们发送一个routingKeyduanxin的消息这个肯定不会发送成功我们通过断点来看看效果效果如下 然后我们常见的就全部整合完成了当然开启了双确认机制虽然我们可以检测到消息投送的结果然后可以针对投送失败的结果进行预警。但是开启了这个操作就必然会对消息的处理效率产生影响。所以还得根据实际业务场景而定是否需要使用这个确认机制。
