消息中间件—SpringBoot下RabbitMQ实战
2020-05-15 16:08:50来源：博客园 阅读 ()
消息中间件—SpringBoot下RabbitMQ实战
消息中间件简介
MQ全称（Message Queue）`又名消息队列，是一种异步通讯的中间件。可以将它理解成邮局，发送者将消息传递到邮局，然后由邮局帮我们发送给具体的消息接收者（消费者），具体发送过程与时间我们无需关心，它也不会干扰我进行其它事情。常见的MQ有kafka、activemq、rocketMQ、rabbitmq等等**
消息中间件的应用场景
消息中间件对比
综上，各种对比之后，有如下建议：
一般的业务系统要引入 MQ，最早大家都用 ActiveMQ，但是现在确实大家用的不多了，没经过大规模吞吐量场景的验证，社区也不是很活跃，所以大家还是算了吧，我个人不推荐用这个了；
后来大家开始用 RabbitMQ，但是确实 erlang 语言阻止了大量的 Java 工程师去深入研究和掌控它，对公司而言，几乎处于不可控的状态，但是确实人家是开源的，比较稳定的支持，活跃度也高；
不过现在确实越来越多的公司会去用 RocketMQ，确实很不错，毕竟是阿里出品，但社区可能有突然黄掉的风险（目前 RocketMQ 已捐给 Apache，但 GitHub 上的活跃度其实不算高）对自己公司技术实力有绝对自信的，推荐用 RocketMQ，否则回去老老实实用 RabbitMQ 吧，人家有活跃的开源社区，绝对不会黄。
所以中小型公司，技术实力较为一般，技术挑战不是特别高，用 RabbitMQ 是不错的选择；大型公司，基础架构研发实力较强，用 RocketMQ 是很好的选择。
如果是大数据领域的实时计算、日志采集等场景，用 Kafka 是业内标准的，绝对没问题，社区活跃度很高，绝对不会黄，何况几乎是全世界这个领域的事实性规范。
安装使用
rabbitmq的windows版本安装，可以参考这个地址： https://blog.csdn.net/hzw19920329/article/details/53156015
Rabbitmq基础概念
Broker：简单来说就是消息队列服务器实体 Exchange：消息交换机，它指定消息按什么规则，路由到哪个队列 Queue：消息队列载体，每个消息都会被投入到一个或多个队列 Binding：绑定，它的作用就是把exchange和queue按照路由规则绑定起来 Routing Key：路由关键字，exchange根据这个关键字进行消息投递 vhost：虚拟主机，一个broker里可以开设多个vhost，用作不同用户的权限分离 producer：消息生产者，就是投递消息的程序 consumer：消息消费者，就是接受消息的程序 channel：消息通道，在客户端的每个连接里，可建立多个channel，每个channel代表一个会话任务
rabbitmq运作流程
上图所示，在整个消息中间件的使用过程中，我们主要配置的是exchange,queue,rooting key三个，下面的代码主要也是根据着三个进行消息的生产和消费。
常用的交换机类型有：fanout、direct、topic、headers，他们特性如下：
fanout：绑定的都发送，忽略路由，不用传routekey
direct：全文匹配--常用
topic：模糊匹配--常用
headers:条件判断，性能很差，不建议使用，我们代码的demo，使用的是direct和topic,后续通过插件的形式，我们还添加了x-delayed-message这个延迟消息队列。
springboot代码实现：
步骤1：pom文件中引入相应的组件
<!--rabbitmq -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
步骤2：部署好mq之后，在application.properties文件中添加rabbitmq相应的配置
spring.rabbitmq.host=129.204.x.xxx
spring.rabbitmq.port=5672
spring.rabbitmq.username=xxx
spring.rabbitmq.password=xxx
spring.rabbitmq.virtual-host=xx
spring.rabbitmq.publisher-confirms=true
spring.rabbitmq.listener.simple.acknowledge-mode=manual步骤3：进行交换机，队列，routingkey的定义和绑定，具体的代码在RabbitConfig文件中。
@Configuration
public class RabbitConfig {
?
//绑定键
public final static String man = "kms.topic.man";
public final static String woman = "kms.topic.woman";
?
/**
* 队列 起名：TestDirectQueue
* @return
*/
@Bean
public Queue DirectQueue() {
// durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
// exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable
// autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
// return new Queue("TestDirectQueue",true,true,false);
//一般设置一下队列的持久化就好,其余两个就是默认false
return new Queue("kms.direct.queue",true);
}
?
/**
* Direct交换机 起名：TestDirectExchange
* @return
*/
@Bean
DirectExchange DirectExchange() {
return new DirectExchange("kms.direct.exchange",true,false);
}
?
/**
* 绑定 将队列和交换机绑定, 并设置用于匹配键：TestDirectRouting
* @return
*/
@Bean
Binding bindingDirect() {
return BindingBuilder.bind(DirectQueue()).to(DirectExchange()).with("kms.direct.routingKey");
}
?
@Bean
public Queue firstTopicQueue() {
return new Queue(RabbitConfig.man);
}
?
@Bean
public Queue secondTopicQueue() {
return new Queue(RabbitConfig.woman);
}
?
/**
* 主题型交换机1
* @return
*/
@Bean
TopicExchange topicExchange() {
return new TopicExchange("kms.topic.exchange");
}
?
/**
* 将firstQueue和topicExchange绑定,而且绑定的键值为topic.man
* 这样只要是消息携带的路由键是topic.man,才会分发到该队列
* @return
*/
@Bean
Binding bindingExchangeMessage() {
return BindingBuilder.bind(firstTopicQueue()).to(topicExchange()).with(man);
}
?
/**
* 将secondQueue和topicExchange绑定,而且绑定的键值为用上通配路由键规则topic.#
* 这样只要是消息携带的路由键是以topic.开头,都会分发到该队列
* @return
*/
@Bean
Binding bindingExchangeMessage2() {
return BindingBuilder.bind(secondTopicQueue()).to(topicExchange()).with("kms.topic.#");
}
?
/**
* TODO:RabbitMQ延迟队列
* @return
*/
@Bean
public Queue delayQueue(){
return QueueBuilder.durable("kms.delay.queue").build();
}
?
@Bean
public CustomExchange delayExchange(){
Map<String,Object> map= Maps.newHashMap();
map.put("x-delayed-type","direct");
return new CustomExchange("kms.delay.exchange","x-delayed-message",true,false,map);
}
?
@Bean
public Binding delayBinding(){
return BindingBuilder.bind(delayQueue()).to(delayExchange()).with("kms.delay.routingKey").noargs();
}
?
?
}
步骤4：生产者创建消息的发布，代码在MqProviderService中,其中留意setup()方法，该方法的作用是为了判断消息是否成功发送到中间件， 消息发送完毕后，则回调此方法 ack代表发送是否成功。
/**
* 发送数据到mq
* @Author: yechongbai
* @Date: 2020/5/11 16:04
* @Copyright: www.zektech.cn
* @since 1.0
*/
@Service
public class MqProviderService extends GlobalResponseHandler {
private final Logger logger = LoggerFactory.getLogger(MqProviderService.class);
?
@Autowired
private RabbitTemplate rabbitTemplate;
?
@PostConstruct
public void setup() {
// 消息发送完毕后，则回调此方法 ack代表发送是否成功
rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
System.out.println("进入回调。。。");
// ack为true，代表MQ已经准确收到消息
if (!ack) {
System.out.println("发送消息到MQ失败");
System.out.println("ConfirmCallback: "+"原因："+cause);
return;
}
try {
System.out.println(correlationData.getId()+":成功发送消息到MQ");
// 修改本地消息表的状态为“已发送”。
//TODO:修改本地表
} catch (Exception e) {
logger.error("警告：修改本地消息表的状态时出现异常", e);
}
?
}
});
}
?
/**
* 发送测试信息--直连交换机
* @return
*/
public String sendDirectMessage(){
String messageId = String.valueOf(UUID.randomUUID());
UserInfoVO userInfoVO=new UserInfoVO();
userInfoVO.setPersonNo(messageId);
userInfoVO.setPersonName("张三");
userInfoVO.setPhoto("111");
//将消息携带绑定键值：TestDirectRouting 发送到交换机TestDirectExchange
System.out.println("推送消息："+ JSON.toJSONString(userInfoVO));
//步骤1.添加订单记录
//前提:消息不能丢失情况需要插入，如果类似微信推送这些不一定需要送到的可以不执行步骤2
// 步骤2.插入推送消息到数据库 步骤1和步骤2需要事务，才能保证后续该消息是否被消费
// CorrelationData 当收到消息回执时，会附带上这个参数
rabbitTemplate.convertAndSend("kms.direct.exchange", "kms.direct.routingKey", userInfoVO,new CorrelationData(messageId));
return buildSuccessResult();
}
?
/**
* 发送测试信息--主题交换机1
* @return
*/
public String sendTopicMessage(){
String messageId = String.valueOf(UUID.randomUUID());
UserInfoVO userInfoVO=new UserInfoVO();
userInfoVO.setPersonNo(messageId);
userInfoVO.setPersonName("主题交换机1");
userInfoVO.setPhoto("111");
//将消息携带绑定键值：TestDirectRouting 发送到交换机TestDirectExchange
System.out.println("推送消息："+ JSON.toJSONString(userInfoVO));
//步骤1.添加订单记录
//前提:消息不能丢失情况需要插入，如果类似微信推送这些不一定需要送到的可以不执行步骤2
// 步骤2.插入推送消息到数据库 步骤1和步骤2需要事务，才能保证后续该消息是否被消费
// CorrelationData 当收到消息回执时，会附带上这个参数
rabbitTemplate.convertAndSend("kms.topic.exchange", "kms.topic.man", JSON.toJSONString(userInfoVO),new CorrelationData(messageId));
return buildSuccessResult();
}
?
/**
* 发送测试信息--主题交换机2
* @return
*/
public String sendTopicMessage2(){
String messageId = String.valueOf(UUID.randomUUID());
UserInfoVO userInfoVO=new UserInfoVO();
userInfoVO.setPersonNo(messageId);
userInfoVO.setPersonName("主题交换机2");
userInfoVO.setPhoto("222");
//将消息携带绑定键值：TestDirectRouting 发送到交换机TestDirectExchange
System.out.println("推送消息："+ JSON.toJSONString(userInfoVO));
//步骤1.添加订单记录
//前提:消息不能丢失情况需要插入，如果类似微信推送这些不一定需要送到的可以不执行步骤2
// 步骤2.插入推送消息到数据库 步骤1和步骤2需要事务，才能保证后续该消息是否被消费
// CorrelationData 当收到消息回执时，会附带上这个参数
rabbitTemplate.convertAndSend("kms.topic.exchange", "kms.topic.woman", JSON.toJSONString(userInfoVO),new CorrelationData(messageId));
return buildSuccessResult();
}
?
/**
* 发送测试信息--延迟队列
* @param ttl
* @return
*/
public String sendDelayMessage(Long ttl){
String messageId = String.valueOf(UUID.randomUUID());
UserInfoVO userInfoVO=new UserInfoVO();
userInfoVO.setPersonNo(messageId);
userInfoVO.setPersonName("张三");
userInfoVO.setPhoto("111");
System.out.println(LocalDateTime.now()+":推送消息："+ JSON.toJSONString(userInfoVO));
rabbitTemplate.convertAndSend("kms.delay.exchange", "kms.delay.routingKey", userInfoVO, new MessagePostProcessor() {
@Override
public Message postProcessMessage(Message message) throws AmqpException {
MessageProperties mp=message.getMessageProperties();
mp.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
mp.setHeader("x-delay",ttl);
System.out.println("延迟队列生产者-发出消息："+JSON.toJSONString(userInfoVO)+",TTL："+ttl);
return message;
}
}, new CorrelationData(messageId));
return buildSuccessResult();
}
}
步骤5：消息接收者---消费消息，新建MqConsumerService，进行消息的监听并且消费。 默认情况下spring-boot-data-amqp是自动ACK机制，就意味着 MQ 会在消息消费完毕后自动帮我们去ACK，这样依赖就存在这样一个问题：如果报错了，消息不会丢失，会无限循环消费，很容易就吧磁盘空间耗完，虽然可以配置消费的次数但这种做法也有失优雅。目前比较推荐的就是我们手动ACK然后将消费错误的消息转移到其它的消息队列中，做补偿处理。由于我们需要手动控制ACK，因此下面监听完消息后需要调用basicAck通知rabbitmq消息已被正确消费，可以将远程队列中的消息删除 ，所以需要手动执行：channel.basicAck(tag,true);
@Component
public class MqConsumerService {
?
/**
* 监听的队列名称 TestDirectQueue
* @param testMessage
* @param channel
* @param tag
* @throws IOException
*/
@RabbitListener(queues = "kms.direct.queue")
public void processDirect(UserInfoVO testMessage,Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
try{
System.out.println("DirectReceiver消费者收到消息 : "+ JSON.toJSONString(testMessage));
//TODO：接收到消息进行业务操作，操作成功，告诉mq该消息已经消费
//执行业务操作，同一个数据不能处理两次，根据业务情况去重，保证幂等性。 （拓展：redis记录处理情况）
// 开启了手工确认机制，如果不加这个，项目重新启动，则改消息会被重新消费
// 异常的话，可以选择让它重新入列，或者丢弃
channel.basicAck(tag,true);
}catch (Exception e){
// 异常情况 :根据需要去： 重发/ 丢弃
// 重发一定次数后， 丢弃， 日志告警,b1:true表示重新入列，false表示不入列
channel.basicNack(tag, false, false);
// 系统 关键数据，永远是有人工干预
}
}
?
/**
* 监听的队列名称 TestDirectQueue
* @param testMessage
* @param channel
* @param tag
* @throws IOException
*/
@RabbitListener(queues = "kms.topic.man")
public void processTopic(String testMessage,Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
try{
JSONObject orderInfo = JSONObject.parseObject(testMessage);
System.out.println("TopicReceiver消费者收到消息 : "+ testMessage);
System.out.println("收到的personNO:"+orderInfo.getString("personNo"));
//TODO：接收到消息进行业务操作
channel.basicAck(tag,true);
}catch (Exception e){
// 异常情况 :根据需要去： 重发/ 丢弃
// 重发一定次数后， 丢弃， 日志告警,b1:true表示重新入列，false表示不入列
channel.basicNack(tag, false, false);
// 系统 关键数据，永远是有人工干预
}
}
?
/**
* 监听的队列名称 TestDirectQueue
* @param testMessage
* @param channel
* @param tag
* @throws IOException
*/
@RabbitListener(queues = "kms.topic.woman")
public void processTopic2(String testMessage,Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
try{
JSONObject orderInfo = JSONObject.parseObject(testMessage);
System.out.println("TopicWomenReceiver消费者收到消息 : "+ testMessage);
System.out.println("收到的personNO:"+orderInfo.getString("personNo"));
channel.basicAck(tag,true);
}catch (Exception e){
// 异常情况 :根据需要去： 重发/ 丢弃
// 重发一定次数后， 丢弃， 日志告警,b1:true表示重新入列，false表示不入列
channel.basicNack(tag, false, false);
// 系统 关键数据，永远是有人工干预
}
}
?
/**
* 监听的队列名称 DelayQueue
* @param testMessage
* @param channel
* @param tag
* @throws IOException
*/
@RabbitListener(queues = "kms.delay.queue")
public void processDelay(UserInfoVO testMessage,Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
try{
System.out.println(LocalDateTime.now()+":DelayQueue消费者收到消息 : "+ JSON.toJSONString(testMessage));
channel.basicAck(tag,true);
}catch (Exception e){
// 异常情况 :根据需要去： 重发/ 丢弃
// 重发一定次数后， 丢弃， 日志告警,b1:true表示重新入列，false表示不入列
channel.basicNack(tag, false, false);
// 系统 关键数据，永远是有人工干预
}
}
?
}
步骤6：进行测试，新增controller，调用消息生产的方法，进行消息消费的查看。
@RestController
@RequestMapping("/remote/mq")
public class TestMqController {
@Autowired
private MqProviderService mqProviderService;
?
@PostMapping("sendDirectMessage")
public String sendDirectMessage(){
mqProviderService.sendDirectMessage();
return "ok";
}
?
@PostMapping("sendTopicMessage")
public String sendTopicMessage(){
mqProviderService.sendTopicMessage();
return "ok";
}
?
@PostMapping("sendTopicMessage2")
public String sendTopicMessage2(){
mqProviderService.sendTopicMessage2();
return "ok";
}
?
@PostMapping("sendDelayMessage")
public String sendDelayMessage(){
mqProviderService.sendDelayMessage(10*1000L);
return "ok";
}
}
执行sendDelayMessage输出结果如下：
2020-05-15T15:16:29.437:推送消息：{"personName":"张三","personNo":"5cd80fa8-42c0-4b70-932f-422ca4167a5a","photo":"111"}
延迟队列生产者-发出消息：{"personName":"张三","personNo":"5cd80fa8-42c0-4b70-932f-422ca4167a5a","photo":"111"},TTL：10000
进入回调。。。
5cd80fa8-42c0-4b70-932f-422ca4167a5a:成功发送消息到MQ
2020-05-15T15:16:39.933:DelayQueue消费者收到消息 : {"personName":"张三","personNo":"5cd80fa8-42c0-4b70-932f-422ca4167a5a","photo":"111"}
原文链接:https://www.cnblogs.com/shuideqing/p/12896308.html
如有疑问请与原作者联系
标签：
版权申明：本站文章部分自网络，如有侵权，请联系：west999com@outlook.com
特别注意：本站所有转载文章言论不代表本站观点，本站所提供的摄影照片，插画，设计作品，如需使用，请与原作者联系，版权归原作者所有
- springboot2配置JavaMelody与springMVC配置JavaMelody 2020-06-11
- 数据源管理 | Kafka集群环境搭建，消息存储机制详解 2020-06-11
- SpringBoot 2.3 整合最新版 ShardingJdbc + Druid + MyBatis 2020-06-11
- 掌握SpringBoot-2.3的容器探针：实战篇 2020-06-11
- nacos~配置中心功能~springboot的支持 2020-06-10
IDC资讯： 主机资讯 注册资讯 托管资讯 vps资讯 网站建设
网站运营： 建站经验 策划盈利 搜索优化 网站推广 免费资源
网络编程： Asp.Net编程 Asp编程 Php编程 Xml编程 Access Mssql Mysql 其它
服务器技术： Web服务器 Ftp服务器 Mail服务器 Dns服务器 安全防护
软件技巧： 其它软件 Word Excel Powerpoint Ghost Vista QQ空间 QQ FlashGet 迅雷
网页制作： FrontPages Dreamweaver Javascript css photoshop fireworks Flash
