RocketMq发送延迟消息

RocketMq发送延迟消息

延迟消息

对于消息中间件来说,producer 将消息发送到mq的服务器上,但并不希望这条消息马上被消费,而是推迟到当前时间节点之后的某个时间点,再将消息投递到 queue 中让 consumer 进行消费。

延迟消息的使用场景很多,一种比较常见的场景就是在电商系统中,订单创建后,会有一个等待用户支付的时间窗口,一般为30分钟,30分钟后 customer 会收到这条订单消息,然后程序去订单表中检查当前这条订单的支付状态,如果是未支付的状态,则自动清理掉,这样就不需要使用定时任务的方式去处理了。

Rocket的延迟消息

RocketMQ 支持定时的延迟消息,但是不支持任意时间精度,仅支持特定的 level,例如定时 5s, 10s, 1m 等。其中,level=0 级表示不延时,level=1 表示 1 级延时,level=2 表示 2 级延时,以此类推。

配置定时的延迟消息方式一:

可以直接在服务器端的 broker.conf 中进行配置,在服务器端(rocketmq-broker端)的属性配置文件中加入以下行,但是这种方式不够灵活,不推荐。

这个配置项配置了从1级开始,各级延时的时间,可以修改这个指定级别的延时时间;

时间单位支持:s、m、h、d,分别表示秒、分、时、天;

默认值就是上面声明的,可手工调整,默认值已够用,不建议修改这个值。

messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h

这个配置项配置了从1级开始,各级延时的时间,可以修改这个指定级别的延时时间;时间单位支持:s、m、h、d,分别表示秒、分、时、天;默认值就是上面声明的,可手工调整;默认值已够用,不建议修改这个值。

配置定时的延迟消息方式二:

在程序中进行指定,这个会在代码中展示。

实际代码演练

我们使用一个controller模拟浏览器调用接口发送一个延迟的消息,为了演示方便发送消息的操作直接放在了controller里面了,实际开发中不要这样做,具体的代码如下所示,添加的maven依赖如下所示:

org.springframework.boot

spring-boot-starter-parent

1.5.8.RELEASE

org.springframework.boot

spring-boot-starter

org.springframework.boot

spring-boot-starter-test

test

org.springframework.boot

spring-boot-starter-web

com.alibaba.rocketmq

rocketmq-client

3.5.8

com.alibaba.rocketmq

rocketmq-all

3.5.8

pom

applicaiton.properties代码如下所示:

spring.application.name=springboot-rocketmq

server.port=8888

###producer

rocketmq.producer.maxMessageSize=4096

rocketmq.producer.sendMsgTimeout=3000

rocketmq.producer.retryTimesWhenSendFailed=2

###consumer

rocketmq.consumer.consumeThreadMin=20

rocketmq.consumer.consumeThreadMax=64

rocketmq.consumer.consumeMessageBatchMaxSize=1

OrderController代码如下所示:

@RestController

@RequestMapping("/api/order")

public class OrderController {

@Autowired

private PayProducer payProducer;

/**

* 发送延迟消息

* @param text

* @return

*/

@GetMapping("/delay")

public Object sendDelayMsg(String text) throws MQClientException, RemotingException, InterruptedException{

Message message = new Message(JmsConfig.TOPIC, "delay_order",("this is a delay message:" + text).getBytes());

//"1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h"

message.setDelayTimeLevel(3);

payProducer.getProducer().send(message, new SendCallback() {

//消息发送成功回调

@Override

public void onSuccess(SendResult sendResult) {

System.out.printf("发送结果=%s, msg=%s ", sendResult.getSendStatus(), sendResult.toString());

}

//消息异常回调

@Override

public void onException(Throwable e) {

e.printStackTrace();

//补偿机制,根据业务情况进行使用,看是否进行重试

}

});

return "send ok";

}

}

payProducer类如下所示:

@Component

public class PayProducer {

private String producerGroup = "pay_producer_group";

private DefaultMQProducer producer;

public PayProducer(){

producer = new DefaultMQProducer(producerGroup);

//生产者投递消息重试次数

producer.setRetryTimesWhenSendFailed(3);

producer.setCreateTopicKey("AUTO_CREATE_TOPIC_KEY");

//指定NameServer地址,多个地址以 ; 隔开

producer.setNamesrvAddr(JmsConfig.NAME_SERVER);

start();

}

public DefaultMQProducer getProducer(){

return this.producer;

}

/**

* 对象在使用之前必须要调用一次,只能初始化一次

*/

public void start(){

try {

this.producer.start();

} catch (MQClientException e) {

e.printStackTrace();

}

}

/**

* 一般在应用上下文,使用上下文监听器,进行关闭

*/

public void shutdown(){

this.producer.shutdown();

}

}

PayConsumer代码如下所示:

@Component

public class PayConsumer {

private DefaultMQPushConsumer consumer;

private String consumerGroup = "pay_consumer_group";

public PayConsumer() throws MQClientException {

consumer = new DefaultMQPushConsumer(consumerGroup);

consumer.setNamesrvAddr(JmsConfig.NAME_SERVER);

consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);

//默认是集群方式,可以更改为广播,但是广播方式不支持重试

consumer.setMessageModel(MessageModel.CLUSTERING);

consumer.subscribe(JmsConfig.TOPIC, "*");

consumer.registerMessageListener( new MessageListenerConcurrently() {

@Override

public ConsumeConcurrentlyStatus consumeMessage(List msgs, ConsumeConcurrentlyContext context) {

MessageExt msg = msgs.get(0);

int times = msg.getReconsumeTimes();

System.out.println("重试次数="+times);

try {

System.out.printf("%s Receive New Messages: %s %n",

Thread.currentThread().getName(), new String(msgs.get(0).getBody()));

String topic = msg.getTopic();

String body = new String(msg.getBody(), "utf-8");

String tags = msg.getTags();

String keys = msg.getKeys();

System.out.println("topic=" + topic + ", tags=" + tags + ", keys=" + keys + ", msg=" + body);

return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;

} catch (Exception e) {

System.out.println("消费异常");

//如果重试2次不成功,则记录,人工介入

if(times >= 2){

System.out.println("重试次数大于2,记录数据库,发短信通知开发人员或者运营人员");

//TODO 记录数据库,发短信通知开发人员或者运营人员

//告诉broker,消息成功

return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;

}

e.printStackTrace();

return ConsumeConcurrentlyStatus.RECONSUME_LATER;

}

}

});

consumer.start();

System.out.println("consumer start ...");

}

}

JmsConfig代码如下所示:

public class JmsConfig {

public static final String NAME_SERVER = "127.0.0.1:9876";

public static final String TOPIC = "DELAY_TOPIC";

}

启动类,启动的输出结果如下所示:

在浏览器中访问我们的controller的地址http://localhost:8888/api/order/delay?text=hello

此时eclipse的输出内容为:

再过10秒钟后的输出结果为:

相关任务

365bet开户送20 国内直播公会机构2023排名第九位:东辰传媒
365bet取款 玩家用数据解析159LG 教你轻松上2200法伤
365bet开户送20 怎么查询Mysql数据库的版本号?(2种常用的方法)
365bet开户送20 教你如何挑选橙子

教你如何挑选橙子

📅 08-26 👁️ 2199