流程描述:
1. producer发消息,设置一个延迟level值.
“设置消息延时 10s 消费”的 Producer 端代码如下：
messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
- Message msg = newMessage(topic, tags, keys, body);
- msg.setDelayTimeLevel(3);
- ...
- SendResult sendResult = getMQProducer().send(msg);
2. broker 保存消息时替换了topic,和queueId(一个level计算得到一个queueId,并将实际的topic和queueId作为properties保存).
3. broker有定时任务(其实是个consumer)消费延迟消息,如果到达延迟时间,将消息取出,改回原来的topic和queueId,放入到commitLog中,然后被真正的消费者.
疑难点:
问: 如何保证rocketMq的offset移动和延迟消息不冲突?
答: rocketMq当消息真正要消费的时候才把消息放到对应的topic中. 中间先保存到其他地方(利用原有的存储,消费机制,自然是一个topic). rocketMq很巧妙的将用户配置的level和queueId进行了一对一映射. 这样就能保证同一个queue下的消息肯定是顺序消费的.
代码细节:
1. producer 发 消息,设置一个level值.
服务端MessageStoreConfig.messageDelayLevel 默认值是
private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";
2. 服务端接受到消息后
MessageConst.PROPERTY_REAL_TOPI
commitLog.putMessage(MessageExtBrokerInner)里,会把配置了延迟level的消息,存到ScheduleMessageService.SCHEDULE_TOPIC(值为SCHEDULE_TOPIC_XXXX) ,
queueId = ScheduleMessageService.delayLevel2QueueId(msg.getDelayTimeLevel());中真正的topic和queueId作为msg暂时存起来.
// Backup real topic, queueIdMessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_TOPIC, msg.getTopic());MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_QUEUE_ID, String.valueOf(msg.getQueueId()));
3.有个定时任务模拟消费者消费该queueScheduleMessageService.DeliverDelayedMessageTimerTask 内,判断是否可消息,可以就取出消息,将topic和quueeId还原,放到commitLog中
PutMessageResult putMessageResult = ScheduleMessageService.this.defaultMessageStore.putMessage(msgInner);//
其中
private static final long FIRST_DELAY_TIME = 1000L; //定时任务第一次启动时延迟时间private static final long DELAY_FOR_A_WHILE = 100L; // 死循环轮训时延迟时间. 又生成一个task,这样避免很多线程一直在执行.比较好的死循环策略.private static final long DELAY_FOR_A_PERIOD = 10000L; // put到commitLog出错时的延迟时间
参考文献:http://www.tuicool.com/articles/aU7JRz7 主要根据该文章的关键字去看对应的源代码,https://github.com/alibaba/RocketMQhttp://blog.csdn.net/fei33423/article/details/51189430