Spring RabbitMQ那些事(2-两种方式实现延时消息订阅)

这篇具有很好参考价值的文章主要介绍了Spring RabbitMQ那些事(2-两种方式实现延时消息订阅)。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

一、序言

业务开发中有很多延时操作的场景,比如最常见的超时订单自动关闭延时异步处理,我们常用的实现方式有:

  • 定时任务轮询(有延时)。
  • 借助Redission的延时队列
  • Redis的key过期事件通知机制(需开启key过期事件通知,对Redis有性能损耗)。
  • RocketMQ中定时消息推送(支持的时间间隔固定,不支持自定义)。
  • RabbitMQ中的死信队列和延迟消息交换机

其中用的最多的也是借助Redisson实现的数据结构延迟队列RabbitMQ中的死信队列来实现,今天我们通过RabbitMQ死信队列和延迟消息交换机(新特性)来实现延时消息推送。


二、死信交换机和消息TTL实现延迟消息

1、死信队列介绍

这种方式主要通过结合消息过期和私信交换机来实现延迟消息推送,首先先了解下哪些消息会进入死信队列:

  • 被消费者nack(negatively acknowleged)的消息。
  • TTL过期后未被消费的消息。
  • 超过队列长度限制后被丢弃的消息。

备注:更多信息请参考RabbitMQ中的 Dead Letter Exchange。

2、代码示例

(1) 死信交换机配置

@Configuration
protected static class DeadLetterExchangeConfig {

	@Bean
	public Queue deadLetterQueue(){
		return QueueBuilder.durable("dead-letter-queue").build();
	}

	@Bean
	public DirectExchange deadLetterExchange() {
		return ExchangeBuilder.directExchange("dead-letter-exchange").build();
	}

	@Bean
	public Binding bindQueueToDeadLetterExchange(Queue deadLetterQueue, DirectExchange deadLetterExchange) {
		return BindingBuilder.bind(deadLetterQueue).to(deadLetterExchange).with("dead-letter-routing-key");
	}

	@Bean
	public Queue normalQueue() {
		return QueueBuilder.durable("normal-queue")
			.deadLetterExchange("dead-letter-exchange")
			.deadLetterRoutingKey("dead-letter-routing-key")
			.build();
	}

}

(2) 消息生产者

@Slf4j
@Component
@RequiredArgsConstructor
public class RabbitMqProducer {

	private final RabbitTemplate rabbitTemplate;
	
	public void sendMsgToDeadLetterExchange(String body, int timeoutInMillSeconds) {
		log.info("开始发送消息到dead letter exchange 消息体:{}, 消息延迟:{}ms, 当前时间:{}", body, timeoutInMillSeconds, LocalDateTime.now());
		MessageProperties messageProperties = MessagePropertiesBuilder.newInstance().setExpiration(String.valueOf(timeoutInMillSeconds)).build();
		Message message = MessageBuilder.withBody(body.getBytes(StandardCharsets.UTF_8)).andProperties(messageProperties).build();
		rabbitTemplate.send("normal-queue", message);
	}
	
}

(3) 消息消费者

@Slf4j
@Component
public class RabbitMqConsumer {

	@RabbitListener(queues = "dead-letter-queue")
	public void handleMsgFromDeadLetterQueue(String msg) {
		log.info("Message received from dead-letter-queue, message body: {}, current time:{}", msg, LocalDateTime.now());
	}
}

3、测试用例

@RestController
@RequiredArgsConstructor
public class RabbitMsgController {

	private final RabbitMqProducer rabbitMqProducer;
	
	@RequestMapping("/exchange/dead-letter")
	public ResponseEntity<String> sendMsgToDeadLetterExchange(String body, int timeout) {
		rabbitMqProducer.sendMsgToDeadLetterExchange(body, timeout);
		return ResponseEntity.ok("消息发送到死信交换机成功");
	}
	
}

浏览器访问http://localhost:8080/exchange/dead-letter?body=hello&timeout=5000,可以看到消息被延迟5s处理。

2023-11-26 11:50:33.041  INFO 19152 --- [nio-8080-exec-7] c.u.r.i.producer.RabbitMqProducer        : 开始发送消息到dead letter exchange 消息体:hello, 消息延迟:5000ms, 当前时间:2023-11-26T11:50:33.041
2023-11-26 11:50:38.054  INFO 19152 --- [ntContainer#4-4] c.u.r.i.consumer.RabbitMqConsumer        : Message received from dead-letter-queue, message body: hello, current time:2023-11-26T11:50:38.054

三、延迟消息交换机实现延迟消息

上面通过消息TTL和死信交换机实现延迟消息的解决方案是由一个叫James Carr的人提出来的,后来RabbitMQ提供了一个开箱即用的解决方案,通过延时消息插件来实现。

该插件以前被当做是试验性产品,但是现在已经可以投产使用了。(PS:2015年4月16号就已经有该插件文档)

Spring AMQP中,同样提供了对该延时消息插件的支持,并且在RabbitMQ 3.6.0版本就已经测试通过。

1、安装延时消息插件

该延时消息插件为社区插件,因此需要自己手动下载安装的RabbMQ版本对应的插件,下载地址:RabbitMQ延时消息插件releases。

我安装的RabbitMQ版本为3.9.9,3.9.0版本的插件对所有3.9.x版本的RabbitMQ都支持。

Spring RabbitMQ那些事(2-两种方式实现延时消息订阅),RabbitMQ,Java面试题,中间件专题,java-rabbitmq,spring,rabbitmq
下载完后把.ez结尾的插件复制RabbitMQ的插件目录下,插件目录为/usr/lib/rabbitmq/plugins

Spring RabbitMQ那些事(2-两种方式实现延时消息订阅),RabbitMQ,Java面试题,中间件专题,java-rabbitmq,spring,rabbitmq
通过命令rabbitmq-plugins enable rabbitmq_delayed_message_exchange安装该插件,通过命令rabbitmq-plugins list查看插件列表,可以看到该延时消息插件已经成功安装。

Spring RabbitMQ那些事(2-两种方式实现延时消息订阅),RabbitMQ,Java面试题,中间件专题,java-rabbitmq,spring,rabbitmq

2、代码示例

(1) 延时消息交换机配置

@Configuration
protected static class DelayedMsgExchangePluginConfig {

	@Bean
	public Queue delayedQueue() {
		return QueueBuilder.durable("delayed-queue").build();
	}

	@Bean
	public DirectExchange delayedExchange() {
		return ExchangeBuilder.directExchange("delayed-exchange").delayed().build();
	}

	@Bean
	public Binding bindDelayedQueueToDelayedChange(Queue delayedQueue, DirectExchange delayedExchange) {
		return BindingBuilder.bind(delayedQueue).to(delayedExchange).with("delayed-routing-key");
	}
}

备注:延时交换机的类型可以为DirectExchage、TopicExcahgeFanoutExchange,这些都支持。

(2) 消息生产者

@Slf4j
@Component
@RequiredArgsConstructor
public class RabbitMqProducer {

	private final RabbitTemplate rabbitTemplate;

	public void sendDelayedMsg(String body, int timeoutInMillSeconds) {
		log.info("开始发送消息到delayed-exchange 消息体:{}, 消息延迟:{}ms, 当前时间:{}", body, timeoutInMillSeconds, LocalDateTime.now());
		MessageProperties messageProperties = new MessageProperties();
		messageProperties.setDelay(timeoutInMillSeconds);
		Message message = MessageBuilder.withBody(body.getBytes(StandardCharsets.UTF_8)).andProperties(messageProperties).build();
		rabbitTemplate.send("delayed-exchange", "delayed-routing-key", message);
	}
}

(3) 消息消费者

@Slf4j
@Component
public class RabbitMqConsumer {

	@RabbitListener(queues = "delayed-queue")
	public void handleMsgFromDelayedQueue(String msg) {
		log.info("Message received from delayed-queue, message body: {}, current time:{}", msg, LocalDateTime.now());
	}
}

3、测试用例

@RestController
@RequiredArgsConstructor
public class RabbitMsgController {

	private final RabbitMqProducer rabbitMqProducer;

	@RequestMapping("/exchange/delayed")
	public ResponseEntity<String> sendMsgToHeadersExchange(String body, int timeout) {
		rabbitMqProducer.sendDelayedMsg(body, timeout);
		return ResponseEntity.ok("消息发送到延迟交换机成功");
	}
	
}

浏览器访问http://localhost:8080/exchange/dead-letter?body=hello&timeout=5000,可以看到消息被延迟5s处理。

2023-11-26 13:02:07.816  INFO 26524 --- [nio-8080-exec-3] c.u.r.i.producer.RabbitMqProducer        : 开始发送消息到delayed-exchange 消息体:Hello, 消息延迟:5000ms, 当前时间:2023-11-26T13:02:07.816
2023-11-26 13:02:12.830  INFO 26524 --- [ntContainer#5-5] c.u.r.i.consumer.RabbitMqConsumer        : Message received from delayed-queue, message body: Hello, current time:2023-11-26T13:02:12.829

四、两种实现方式优缺点

1、延时消息插件

  • 优点:配置更加简单,少配置1个过期消息接收队列,且语义更明确,容易定位消息出入口。
  • 缺点:延时消息插件对RabbitMQ版本有要求,只有RabbitMQ 3.8.x及以上版本支持。

2、TLL&死信交换机

  • 优点:基本适用于所有RabbitMQ版本。
  • 缺点:配置相对来说复杂一些,还有就是我们最开始提到的,不只是TTL过期的消息才会进入死信队列,还有超出队列限制nack的消息也会进入死信队列,触发的条件没那么纯粹。

Spring RabbitMQ那些事(2-两种方式实现延时消息订阅),RabbitMQ,Java面试题,中间件专题,java-rabbitmq,spring,rabbitmq文章来源地址https://www.toymoban.com/news/detail-765301.html

到了这里,关于Spring RabbitMQ那些事(2-两种方式实现延时消息订阅)的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • RabbitMQ实现延迟消息,RabbitMQ使用死信队列实现延迟消息,RabbitMQ延时队列插件

    假设有一个业务场景:超过30分钟未付款的订单自动关闭,这个功能应该怎么实现? RabbitMQ使用死信队列,可以实现消息的延迟接收。 队列有一个消息过期属性。就像丰巢超过24小时就收费一样,通过设置这个属性,超过了指定事件的消息将会被丢弃。 这个属性交:x-message

    2024年02月13日
    浏览(49)
  • 使用StreamBridge实现RabbitMq 消息收发 && ack确认 && 延时消息

    下载地址:link 1.下载完成放到rabbitmq安装目录plugins下 2.执行命令启用插件 3.重启mq Exchanges - add a new exchange - type 出现x-delayed-message即安装成功

    2024年02月11日
    浏览(30)
  • .Net 6 下WorkerService+RabbitMq实现消息的异步发布订阅

            近期项目里有需要用到RabbitMq实现一些业务,学习整理之后在此记录一下,如有问题或者不对的地方,欢迎留言指正。 注意: 多线程消息发布时,应避免多个线程使用同一个IModel实例,必须保证Imodel被一个线程独享,如果必须要多个线程访问呢一个实例的话,则可以

    2024年01月18日
    浏览(36)
  • Spring整合RabbitMQ-配制文件方式-3-消息拉模式

    拉消息的消费者 spring-rabbit.xml 当启动消费者后,便可获取到发送至队列的消息 检查队列的消息的情况: 经过检查确认,发现消息已经被消费了。 至此拉模式的消费者完成。

    2024年02月09日
    浏览(31)
  • Spring整合RabbitMQ-配制文件方式-1-消息生产者

    Spring-amqp是对AMQP的一些概念的一些抽象,Spring-rabbit是对RabbitMQ操作的封装实现。 主要有几个核心类 RabbitAdmin 、 RabbitTemplate 、 SimpleMessageListenerContainer 等 RabbitAdmin 类完成对Exchange、Queue、Binding的操作,在容器中管理 了 RabbitAdmin 类的时候,可以对Exchange、Queue、Binding进行自动声

    2024年02月09日
    浏览(32)
  • 【Spring Boot】集成Kafka实现消息发送和订阅

    最近忙着搞低代码开发,好久没新建spring项目了,结果今天心血来潮准备建个springboot项目 注意Type选Maven,java选8,其他默认 点下一步后完成就新建了一个spring boot项目,配置下Maven环境,主要是settings.xml文件,里面要包含阿里云仓库,不然可能依赖下载不下来 在maven配置没问

    2024年02月09日
    浏览(32)
  • 使用Spring Boot和Kafka实现消息发送和订阅

    最近忙着搞低代码开发,好久没新建spring项目了,结果今天心血来潮准备建个springboot项目 注意Type选Maven,java选8,其他默认 点下一步后完成就新建了一个spring boot项目,配置下Maven环境,主要是settings.xml文件,里面要包含阿里云仓库,不然可能依赖下载不下来 在maven配置没问

    2024年02月11日
    浏览(29)
  • 使用Spring Boot和Kafka实现消息订阅和发送

    最近忙着搞低代码开发,好久没新建spring项目了,结果今天心血来潮准备建个springboot项目 注意Type选Maven,java选8,其他默认 点下一步后完成就新建了一个spring boot项目,配置下Maven环境,主要是settings.xml文件,里面要包含阿里云仓库,不然可能依赖下载不下来 在maven配置没问

    2024年02月11日
    浏览(27)
  • RabbitMq应用延时消息

    一.建立绑定关系 二.建立生产者 1.消息实体 三.建立消费者 四.测试类测试 五.效果如图所示

    2024年02月12日
    浏览(29)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包