七 延时消息
延时消息:Producer 将消息发送到消息队列 RocketMQ 服务端,但并不期望这条消息立马投递(被消费者消费),而是延迟一定时间后才投递到 Consumer 进行消费,该消息即延时消息。
适用场景:消息生产和消费有时间窗口要求:比如在电商交易中超时未支付关闭订单的场景,在订单创建时向RocketMQ发送一条延时消息。这条消息将会在 30 分钟以后投递给消费者,消费者收到此消息后需要判断对应的订单是否已完成支付。 如支付未完成,则关闭订单。如已完成支付则忽略。
使用案例:Apache RocketMQ目前只支持固定精度的定时消息,因为如果要支持任意的时间精度,在 Broker 层面,必须要做消息排序,如果再涉及到持久化,那么消息排序要不可避免的产生巨大性能开销。(RocketMQ的商业版本Aliware MQ提供了任意时刻的定时消息功能,Apache的RocketMQ并没有,阿里并没有开源)
Apache RocketMQ发送延时消息是设置在每一个消息体上的,在创建消息时设定一个延时时间长度,消息将从当前发送时间点开始延迟固定时间之后才开始投递。
延迟消息的level,区分18个等级:level为1,表示延迟1秒后消费;level为2表示延迟5秒后消费;level为3表示延迟10秒后消费;以此类推;最大level为18表示延迟2个小时消费。具体标识如下:
level |
1 |
2 |
3 |
4 |
5 |
6 |
7 |
8 |
9 |
延迟 |
1s |
5s |
10s |
30s |
1m |
2m |
3m |
4m |
5m |
level |
10 |
11 |
12 |
13 |
14 |
15 |
16 |
17 |
18 |
延迟 |
6m |
7m |
8m |
9m |
10m |
20m |
30m |
1h |
2h |
代码演示生产者
package com.neu.rocketmq.example.delay;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
/**
* Created with IntelliJ IDEA.
*
* @Author: yqq
* @Date: 2023/06/20/15:04
* @Description:
*/
public class DelayProducer {
public static void main(String[] args) throws Exception {
//实例化一个生产者
DefaultMQProducer producer = new DefaultMQProducer("ScheduledProducer");
//设置NameServer地址
producer.setNamesrvAddr("node1:9876");
//启动生产者
producer.start();
for (int i = 0; i
代码演示消费者
package com.neu.rocketmq.example.delay;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.List;
/**
* Created with IntelliJ IDEA.
*
* @Author: yqq
* @Date: 2023/06/20/15:11
* @Description:
*/
public class DelayConsumer {
public static void main(String[] args) throws Exception {
//实例化消费者
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ScheduledConsumer");
//指定NameServer地址
consumer.setNamesrvAddr("node1:9876");
//订阅Topic
consumer.subscribe("ScheduledTopic","*");
//注册监听者
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List msgs, ConsumeConcurrentlyContext context) {
for (MessageExt msg : msgs){
//打印延时消息
System.out.println("Receive message[msgId=" + msg.getMsgId() + "]" + (
msg.getStoreTimestamp() - msg.getBornTimestamp()
) + "ms later");
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
//启动消费者
consumer.start();
}
}
服务器托管,北京服务器托管,服务器租用 http://www.fwqtg.net
机房租用,北京机房租用,IDC机房托管, http://www.fwqtg.net
Fast Framework 作者 Mr-zhong 代码改变世界…. 一、前言 Fast Framework 基于NET6.0 封装的轻量级 ORM 框架 支持多种数据库 SqlServer Oracle MySql PostgreSql Sqlite …