RocketMQ延时消息

RocketMQ延时消息

RocketMQ提供一种延时消息的解决方案,就是在特定的时间到了,消息才会被投递出去供consumer消费。

总体来说可以满足简单的场景是,但是需要注意的是延时的时间是需要按照内置的延时级别去配置的,而不是随意设置消息的延时时间。

如果想不受延时级别的约束,可以参考:http://blog.seoui.com/2017/08/19/delayqueue/

默认的延迟级别

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

这个配置下标从1开始,比如级别2是延时5秒、级别5是延时1分钟。默认配置在不满足需求的情况下,可以在broker配置文件加入messageDelayLevel参数覆盖默认的延时级别配置。

RocketMQ发送延时消息

和普通的消息不同之处在于Producer在发送消息的时候,需要设置message.setDelayTimeLevel();延迟级别方法,其他参数和消费端的写法并无不同之处。

Producer

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
public class ScheduledMessageProducer {
public static void main(String[] args) throws Exception {
DefaultMQProducer producer =
new DefaultMQProducer("ExampleProducerGroup");
producer.start();
int totalMessagesToSend = 100;
for (int i = 0; i < totalMessagesToSend; i++) {
Message message = new Message("TestTopic",
("Hello scheduled message " + i).getBytes());
//延时的级别为3 对应的时间为10s 就是发送后延时10S在把消息投递出去
message.setDelayTimeLevel(3);
producer.send(message);
}
producer.shutdown();
}
}

Consumer

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
public class ScheduledMessageConsumer {
public static void main(String[] args) throws Exception {
DefaultMQPushConsumer consumer =
new DefaultMQPushConsumer("ExampleConsumer");
consumer.subscribe("TestTopic", "*");
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messages,
ConsumeConcurrentlyContext context) {
for (MessageExt message : messages) {
System.out.println("ok!");
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
consumer.start();
}
}

Powered by AppBlog.CN     浙ICP备14037229号

Copyright © 2012 - 2021 APP开发技术博客 All Rights Reserved.

访客数 : | 访问量 :