#RabbitMQ

之前在写Spring Cloud Stream专题内容的时候,特地介绍了一下如何使用RabbitMQ的延迟消息来实现定时任务。最近正好因为开发碰到了使用过程中发现,延迟消息没有效果,消息直接就被消费了的情况。因此就继续深入研究了一下问题原因,在此记录下来,给碰到类似问题的童鞋们参考。 问题定位因为不是所有的消息都出现了没有延迟消息效果的因素,通过有问题的消息特征,大致猜测可能是延迟时间过长导致了消息延迟失败。为了验证这个原因,先拿之前文章中的例子,来测试一下延迟时间是否与问题直接相关。 对之前的延迟消息使用样例(文末的Git仓库中可以获取完整代码)接口做一下微改,增加了一个请求参数delay来控制延迟时间: @GetMapping("/sendMessage")public String messageWithMQ(@RequestParam String message, @Req...

应用场景我们在使用一些开源调度系统(比如:elastic-job等)的时候,对于任务的执行时间通常都是有规律性的,可能是每隔半小时执行一次,或者每天凌晨一点执行一次。然而实际业务中还存在另外一种定时任务,它可能需要一些触发条件才开始定时,比如:编写博文时候,设置2小时之后发送。对于这些开始时间不确定的定时任务,我们也可以通过Spring Cloud Stream来很好的处理。 为了实现开始时间不确定的定时任务触发,我们将引入延迟消息的使用。RabbitMQ中提供了关于延迟消息的插件,所以本文就来具体介绍以下如何利用Spring Cloud Stream以及RabbitMQ轻松的处理上述问题。 动手试试插件安装关于RabbitMQ延迟消息的插件介绍可以查看官方网站:https://www.rabbitmq.com/blog/2015/04/16/scheduling-messages-...

应用场景之前我们已经通过《Spring Cloud Stream消费失败后的处理策略(一):自动重试》一文介绍了Spring Cloud Stream默认的消息重试功能。本文将介绍RabbitMQ的binder提供的另外一种重试功能:重新入队。 动手试试准备一个会消费失败的例子,可以直接沿用前文的工程,也可以新建一个,然后创建如下代码的逻辑: @EnableBinding(TestApplication.TestTopic.class)@SpringBootApplicationpublic class TestApplication { public static void main(String[] args) { SpringApplication.run(TestApplication.class, args); } ...

应用场景前两天我们已经介绍了两种Spring Cloud Stream对消息失败的处理策略: 自动重试:对于一些因环境原因(如:网络抖动等不稳定因素)引发的问题可以起到比较好的作用,提高消息处理的成功率。 自定义错误处理逻辑:如果业务上,消息处理失败之后有明确的降级逻辑可以弥补的,可以采用这种方式,但是2.0.x版本有Bug,2.1.x版本修复。 那么如果代码本身存在逻辑错误,无论重试多少次都不可能成功,也没有具体的降级业务逻辑,之前在深入思考中讨论过,可以通过日志,或者降级逻辑记录的方式把错误消息保存下来,然后事后分析、修复Bug再重新处理。但是很显然,这样做非常原始,并且太过笨拙,处理复杂度过高。所以,本文将介绍利用中间件特性来便捷地处理该问题的方案:使用RabbitMQ的DLQ队列。 动手试试准备一个会消费失败的例子,可以直接沿用前文的工程。也可以新建一个,然后创建如下代...

最近收到好几个类似的问题:使用Spring Cloud Stream操作RabbitMQ或Kafka的时候,出现消息重复消费的问题。通过沟通与排查下来主要还是用户对消费组的认识不够。其实,在之前的博文以及《Spring Cloud微服务实战》一书中都有提到关于消费组的概念以及作用。 那么什么是消费组呢?为什么要用消费组?它解决什么问题呢?摘录一段之前博文的内容,来解答这些疑问: 通常在生产环境,我们的每个服务都不会以单节点的方式运行在生产环境,当同一个服务启动多个实例的时候,这些实例都会绑定到同一个消息通道的目标主题(Topic)上。默认情况下,当生产者发出一条消息到绑定通道上,这条消息会产生多个副本被每个消费者实例接收和处理(出现上述重复消费问题)。但是有些业务场景之下,我们希望生产者产生的消息只被其中一个实例消费,这个时候我们需要为这些消费者设置消费组来实现这样的功能。 详细...

有很多网友留言:公司要做消息中间件选型,该如何选?你觉得哪个比较好?消息选型的确是一个大论题,实则说来话长的事情又如何长话短说。对此笔者专门撰稿一篇内功心法:如何看待消息中间件的选型,不过这篇只表其意未表其行,为了弥补这种缺陷,笔者最近特意重新撰稿一篇,以供参考。温馨提示:本文一万多字,建议先马(关注)后看。 一、前言消息队列中间件(简称消息中间件)是指利用高效可靠的消息传递机制进行与平台无关的数据交流,并基于数据通信来进行分布式系统的集成。通过提供消息传递和消息排队模型,它可以在分布式环境下提供应用解耦、弹性伸缩、冗余存储、流量削峰、异步通信、数据同步等等功能,其作为分布式系统架构中的一个重要组件,有着举足轻重的地位。 目前开源的消息中间件可谓是琳琅满目,能让大家耳熟能详的就有很多,比如ActiveMQ、RabbitMQ、Kafka、RocketMQ、ZeroMQ等。不管选择其...

先回顾一下,在之前的Spring Cloud Config的介绍中,我们还留了一个悬念:如何实现对配置信息的实时更新。虽然,我们已经能够通过/refresh接口和Git仓库的Web Hook来实现Git仓库中的内容修改触发应用程序的属性更新。但是,若所有触发操作均需要我们手工去维护Web Hook中的应用位置的话,这随着系统的不断扩张,会变的越来越难以维护,而消息代理中间件是解决该问题最为合适的方案。是否还记得我们在介绍消息代理中的特点时有提到过这样一个功能:消息代理中间件可以将消息路由到一个或多个目的地。利用这个功能,我们就能完美的解决该问题,下面我们来说说Spring Cloud Bus中的具体实现方案。 在《Spring Boot中使用RabbitMQ》一文中,我们已经介绍了关于消息代理、AMQP协议以及RabbitMQ的基础知识和使用方法。下面我们开始具体介绍Spring C...

很久没有写Spring Boot的内容了,正好最近在写Spring Cloud Bus的内容,因为内容会有一些相关性,所以先补一篇关于AMQP的整合。 Message Broker与AMQP简介Message Broker是一种消息验证、传输、路由的架构模式,其设计目标主要应用于下面这些场景: 消息路由到一个或多个目的地 消息转化为其他的表现方式 执行消息的聚集、消息的分解,并将结果发送到他们的目的地,然后重新组合相应返回给消息用户 调用Web服务来检索数据 响应事件或错误 使用发布-订阅模式来提供内容或基于主题的消息路由 AMQP是Advanced Message Queuing Protocol的简称,它是一个面向消息中间件的开放式标准应用层协议。AMQP定义了这些特性: 消息方向 消息队列 消息路由(包括:点到点和发布-订阅模式) 可靠性 安全性 RabbitMQ本文要...