首页 > 其他分享 >MQ系列14:MQ如何做到消息延时处理

MQ系列14:MQ如何做到消息延时处理

时间:2023-09-06 12:56:01浏览次数:40  
标签:系列 14 发送 MQ 消息 延时 RocketMQ

MQ系列1:消息中间件执行原理
MQ系列2:消息中间件的技术选型
MQ系列3:RocketMQ 架构分析
MQ系列4:NameServer 原理解析
MQ系列5:RocketMQ消息的发送模式
MQ系列6:消息的消费
MQ系列7:消息通信,追求极致性能
MQ系列8:数据存储,消息队列的高可用保障
MQ系列9:高可用架构分析
MQ系列10:如何保证消息幂等性消费
MQ系列11:如何保证消息可靠性传输
MQ系列12:如何保证消息顺序性
MQ系列13:消息大量堆积如何解决

1 背景

在互联网业务的实际应用场景中,消息的延时处理是非常必要的。例如,在金融交易系统中,某些交易的确认可能需要一段时间才能完成。又如,在物流跟踪系统中,货物的运输状态需要一段时间才能更新。而MQ作为中间件的角色专门来处理消息媒介,实际也具备了使用消息的延时处理来保证信息的及时性的能力。
这边举两个具体的例子:

  • 火车票订购,提交了订单就把车票给占位了,这时候可以发送一个延时确认的消息,15m 未付款,就要把该车票释放,这样其他人就可以购买了。
  • 购买电影票,可以发送一个核销检查消息,在电影开场前15分钟就无法退票了。

既然消息延迟处理的使用场景这么常见,那我们就要详细来分析下怎么使用MQ来实现,这边以RocketMQ为技术选型。

2 消息延时处理原理

RocketMQ的消息延时处理是通过预定义的消息延时级别和延时队列来实现的。在发送消息时,生产者可以设置一个延时级别,该消息将会被延迟一段时间后才能被消费者消费。RocketMQ默认提供了18个延时级别,每个级别对应不同的延迟时间。

所以延时时间并不是随意指定的,Rocket源码中指定的18种等级如下:

// org/apache/rocketmq/store/config/MessageStoreConfig.java 的第198行
private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";
  • RocketMq不支持任意时间延时,需设置固定的延时等级,从1s到2h分别对应着等级1到18
  • 可以使用setDelayTimeLevel(int level) 方法设置延时等级,level 从 0 开始

在RocketMQ中,每个Broker都设置了一个延时队列,用于存储延时消息。当消息的延时时间到达时,该消息将会被自动转移到普通的消息队列中,等待消费者的消费。这种方式可以有效地避免因为网络延迟或者消费者处理速度慢而导致消息的延迟。

image

3 消息延时处理实战

使用RocketMQ的消息延时处理非常简单。在发送消息时,生产者只需要设置一个延时级别,然后将消息发送到RocketMQ即可。例如:

public class DelayProducerApplication {
    public static void main(String[] args) throws InterruptedException, RemotingException, MQClientException, MQBrokerException , UnsupportedEncodingException {
        // 1、创建生产者producer,并指定生产者组名为 example_group_name
        DefaultMQProducer producer = new DefaultMQProducer("example_group_name");  
        // 2、指定NameServer的地址,以获取Broker路由地址
        producer.setNamesrvAddr("127.0.0.1:9876");
        // 3、启动生产者producer
        producer.start();
        // 4、创建消息,并指定Topic,Tag和消息体
        Message msg = new Message("example_topic","example_key", "试一试延迟30s发送的消息".getBytes("UTF-8"));
        // 5、设置延时等级3,对应30s,所以这个消息在30秒之后发送
        msg.setDelayTimeLevel(3);
        // 6、发送消息到一个Broker
        SendResult sendResult = producer.send(msg);
        // 7、通过sendResult返回消息是否成功送达
        System.out.printf("%s%n", sendResult);
        // 8、如果不再发送消息,关闭生产者Producer
        producer.shutdown();
    }
}

image

在上述代码中,我们首先创建了一个生产者,然后指定了NameServer的地址,并启动了生产者。接着,我们创建了一个延时级别为3的消息,即该消息将会被延迟30秒后才能发送并被消费者消费。最后,我们发送了该消息,并关闭了生产者。

4 消息延时的优化

虽然RocketMQ的消息延时处理功能已经非常强大,但是在实际应用中,我们可能还需要根据自己的业务需求进行一些优化。以下是一些可能的优化方式:

  • 调整延时队列的大小。在RocketMQ中,每个Broker都只有一个延时队列,队列太小可能导致一些延时消息被miss。可以根据实际需求调整延时队列的大小。
  • 使用多个消费者来消费同一主题的消息。在RocketMQ中,可能有批量执行被设置了同样的延迟时间,这个就存在了一些风险,类似缓存的批量过期一样,稍有不慎,可能会击穿数据库。如果只有一个消费者来消费该主题的消息,可能会导致该消费者的处理速度不够快,从而影响到消息的及时性。我们可以根据实际需求增加消费者数量,以提高消息的处理速度。
  • 调整RocketMQ的配置参数。RocketMQ提供了一些配置参数,可以用来调整其性能和可靠性。我们可以根据实际需求调整这些参数,以优化消息的延时处理效果。

总之,RocketMQ的消息延时处理功能非常强大,可以满足许多实际应用场景的需求。在实际应用中,我们可以根据自己的业务需求进行一些优化,以进一步提高消息的及时性和可靠性。

5 总结

本文我们介绍了RocketMQ如何使用消息延时来处理特殊的业务场景。除了上述的方法之外,我们还有一些其他方法,比如:

  • 定时发送消息。在定时发送中,生产者可以指定一个未来的时间戳,在该时间戳到达时,该消息将会被发送到Broker。RocketMQ内部会维护一个定时任务,每隔一段时间检查一次待定时发送的消息,并判断是否到达了指定的时间戳。如果到达了指定的时间戳,该消息将会被发送到Broker。
  • 自建环形队列来实现“延时消息” ,参考这篇:1分钟实现“延迟消息”功能,写的不错

标签:系列,14,发送,MQ,消息,延时,RocketMQ
From: https://www.cnblogs.com/wzh2010/p/15888549.html

相关文章

  • typecho_v1.0-14.10.10_反序列化漏洞复现
    目录漏洞利用GetShell下载链接:https://pan.baidu.com/s/1z0w7ret-uXHMuOZpGYDVlw提取码:lt7aTypecho-反序列化漏洞大佬博客Typechoinstall.php存在的反序列化漏洞首页漏洞点:/install.php?finish=漏洞利用漏洞利用脚本phpinfo()信息<?php//typecho_1.0(14.10.......
  • 深入了解Mosquitto:MQTT消息的用法和介绍
    MQTT(MessageQueuingTelemetryTransport)是一种轻量级的消息传输协议,在物联网领域得到广泛应用。Mosquitto是一个流行的MQTT消息代理实现,它允许设备和应用之间进行高效的消息通信。本文将介绍Mosquitto的基本用法和特点,并提供代码演示。什么是Mosquitto?Mosquitto是一个开源的MQTT......
  • C++14新特性
    这篇文章介绍下C++14的新特性。 函数返回值类型推导 C++14对函数返回类型推导规则做了优化,先看一段代码: #include<iostream>usingnamespacestd;autofunc(inti){returni;}intmain(){cout<<func(4)<<endl;return0;} 使用C++11编......
  • RabbitMQ安装 - 二进制安装(CentOS7 RabbitMQ 3.11.3)
    Linux系统-部署-运维系列导航 RabbitMQ介绍RabbitMQ是使用Erlang语言开发的基于AMQP标准的开源实现,用于在分布式系统中存储转发消息,在易用性、扩展性、高可用性等方面表现不错 RabbitMQ的特点1、保证可靠性(Reliability):使用持久化、传输确认、发布确认等机制2、灵活的路......
  • redis cluster集群安装(CentOS7 + redis 5.0.14)
    Linux系统-部署-运维系列导航 rediscluster介绍redis最开始使用主从模式做集群,若master宕机需要手动配置slave转为master;后来为了高可用提出来哨兵模式,该模式下有一个哨兵监视master和slave,若master宕机可自动将slave转为master,但它也有一个问题,就是不能动态扩充;所以在3.x提......
  • ubuntu服务器上部署EMQX SLL
    1.    安装MQTT的简易服务器集成环境EMQX$wgethttps://www.emqx.com/zh/downloads/broker/5.1.6/emqx-5.1.6-ubuntu18.04-amd64.deb$sudoaptinstall./emqx-5.1.6-ubuntu18.04-amd64.deb$sudosystemctlstartemqx具体版本下载请访问:www.emqx.com/zh/try?product=bro......
  • Linux rocketmq单机测试部署
    一.环境说明对于RocketMQ4.3.0版本,官方要求环境如下,其中Git用于从GitHub获取源码,没有安装也没关系,可以直接下载推荐的流程是:Linux系统上安装Git工具、Maven、JavaJDKGit工具用于直接从GitHub获取RocketMQ项目源码下载到Linux系统上然后Maven将RocketMQ源码......
  • MQ消息中间件之RabbitMQ
    1.MQ知识扩充1.1什么是MQ消息中间件MQ全称MessageQueue(消息队列),是在消息的传输过程中保存消息的容器。它是应用程序和应用程序之间的通信方法1.2为什么使用MQ【注:在项目中,可将一些无需即时返回且耗时的操作提取出来,进行异步处理,而这种异步处理的方式大大的节省了服务器......
  • rocketmq的转发模型
    1.模型 2.消息模型RocketMQ主要由Producer、Broker、Consumer三部分组成,其中Producer负责生产消息,Consumer负责消费消息,Broker负责存储消息。Broker在实际部署过程中对应一台服务器,每个Broker可以存储多个Topic的消息,每个Topic的消息也可以分片存储于不同......
  • 14.MySQL数据库设计详解
    MySQL数据库设计需要根据具体的业务需求和数据模型来制定,以下是一个示例数据库设计,包括创建表、定义索引、外键关系和示例数据插入的MySQL代码。这个示例涵盖了一个简单的电子商务系统,包括用户、产品和订单数据。--创建用户表CREATETABLEusers(user_idINTAUTO_INCREM......