生产者把消息发送到消息队列中以后,并不期望被立即消费,而是等待指定时间后才可以被消费者消费,这类消息通常被称为延迟消息。延迟消息的应用场景其实是非常的广泛,比如以下的场景:
rabbitmq-plugins enable rabbitmq_delayed_message_exchange
x-delayed-message
类型的交换器机:Map<String, Object> args = new HashMap<>();
args.put("x-delayed-type", "direct");
channel.exchangeDeclare("one-more-exchange", "x-delayed-message", true, false, args);
x-delay
,表示延迟的毫秒数:byte[] messageBodyBytes = "This is a delayed message".getBytes();
AMQP.BasicProperties.Builder props = new AMQP.BasicProperties.Builder();
headers = new HashMap<String, Object>();
headers.put("x-delay", 5000);
props.headers(headers);
channel.basicPublish("one-more-exchange", "", props.build(), messageBodyBytes);
<broker xmlns="http://activemq.apache.org/schema/core" schedulerSupport="true">
</broker>
MessageProducer producer = session.createProducer(destination);
TextMessage message = session.createTextMessage("This is a delayed message");
message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, 60 * 1000);
producer.send(message);
MessageProducer producer = session.createProducer(destination);
TextMessage message = session.createTextMessage("This is a delayed message");
message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, 60 * 1000);
message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_PERIOD, 10 * 1000);
message.setIntProperty(ScheduledMessage.AMQ_SCHEDULED_REPEAT, 4);
producer.send(message);
MessageProducer producer = session.createProducer(destination);
TextMessage message = session.createTextMessage("This is a delayed message");
message.setStringProperty(ScheduledMessage.AMQ_SCHEDULED_CRON, " 0 3 * * *");
producer.send(message);
MessageProducer producer = session.createProducer(destination);
TextMessage message = session.createTextMessage("test msg");
message.setStringProperty(ScheduledMessage.AMQ_SCHEDULED_CRON, "0 * * * *");
message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, 60 * 1000);
message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_PERIOD, 10 * 1000);
message.setIntProperty(ScheduledMessage.AMQ_SCHEDULED_REPEAT, 4);
producer.send(message);
public class Consumer {
public static void main(String[] args) throws MQClientException {
SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss.SSS");
// 实例化消费者
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("OneMoreGroup");
// 设置NameServer的地址
consumer.setNamesrvAddr("localhost:9876");
// 订阅一个或者多个Topic,以及Tag来过滤需要消费的消息
consumer.subscribe("OneMoreTopic", "*");
// 注册回调实现类来处理从broker拉取回来的消息
consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
System.out.printf("%s %s Receive New Messages:%n"
, sdf.format(new Date())
, Thread.currentThread().getName());
for (MessageExt msg : msgs) {
System.out.printf("\tMsg Id: %s%n", msg.getMsgId());
System.out.printf("\tBody: %s%n", new String(msg.getBody()));
}
// 标记该消息已经被成功消费
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
});
// 启动消费者实例
consumer.start();
System.out.println("Consumer Started.");
}
}
public class DelayProducer {
public static void main(String[] args) throws Exception {
SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss.SSS");
// 实例化消息生产者Producer
DefaultMQProducer producer = new DefaultMQProducer("OneMoreGroup");
// 设置NameServer的地址
producer.setNamesrvAddr("localhost:9876");
// 启动Producer实例
producer.start();
Message msg = new Message("OneMoreTopic"
, "DelayMessage", "This is a delay message.".getBytes());
//"1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h"
//设置消息延迟级别为3,也就是延迟10s。
msg.setDelayTimeLevel(3);
// 发送消息到一个Broker
SendResult sendResult = producer.send(msg);
// 通过sendResult返回消息是否成功送达
System.out.printf("%s Send Status: %s, Msg Id: %s %n"
, sdf.format(new Date())
, sendResult.getSendStatus()
, sendResult.getMsgId());
// 如果不再发送消息,关闭Producer实例。
producer.shutdown();
}
}
10:37:14.992 Send Status: SEND_OK, Msg Id: C0A8006D5AB018B4AAC216E0DB690000
10:37:25.026 ConsumeMessageThread_1 Receive New Messages:
Msg Id: C0A8006D5AB018B4AAC216E0DB690000
Body: This is a delay message.
// 延迟级别大于0,就是延时消息
if (msg.getDelayTimeLevel() > 0) {
// 判断当前延迟级别,如果大于最大延迟级别,
// 就设置当前延迟级别为最大延迟级别。
if (msg.getDelayTimeLevel() > this.defaultMessageStore
.getScheduleMessageService().getMaxDelayLevel()) {
msg.setDelayTimeLevel(this.defaultMessageStore
.getScheduleMessageService().getMaxDelayLevel());
}
// 获取延迟消息的主题,
// 其中RMQ_SYS_SCHEDULE_TOPIC的值为SCHEDULE_TOPIC_XXXX
topic = TopicValidator.RMQ_SYS_SCHEDULE_TOPIC;
// 根据延迟级别获取延迟消息的队列Id,
// 队列Id其实就是延迟级别减1
queueId = ScheduleMessageService.delayLevel2QueueId(msg.getDelayTimeLevel());
// 备份真正的主题和队列Id
MessageAccessor.putProperty(msg
, MessageConst.PROPERTY_REAL_TOPIC, msg.getTopic());
MessageAccessor.putProperty(msg
, MessageConst.PROPERTY_REAL_QUEUE_ID, String.valueOf(msg.getQueueId()));
msg.setPropertiesString(MessageDecoder.messageProperties2String(msg.getProperties()));
// 设置延时消息的主题和队列Id
msg.setTopic(topic);
msg.setQueueId(queueId);
}
load
方法。最后,再执行ScheduleMessageService的start
方法:public void start() {
// 使用AtomicBoolean确保start方法仅有效执行一次
if (started.compareAndSet(false, true)) {
this.timer = new Timer("ScheduleMessageTimerThread", true);
// 遍历所有延迟级别
for (Map.Entry<Integer, Long> entry : this.delayLevelTable.entrySet()) {
// key为延迟级别
Integer level = entry.getKey();
// value为延迟级别对应的毫秒数
Long timeDelay = entry.getValue();
// 根据延迟级别获得对应队列的偏移量
Long offset = this.offsetTable.get(level);
// 如果偏移量为null,则设置为0
if (null == offset) {
offset = 0L;
}
if (timeDelay != null) {
// 为每个延迟级别创建定时任务,
// 第一次启动任务延迟为FIRST_DELAY_TIME,也就是1秒
this.timer.schedule(new DeliverDelayedMessageTimerTask(level, offset), FIRST_DELAY_TIME);
}
}
// 延迟10秒后每隔flushDelayOffsetInterval执行一次任务,
// 其中,flushDelayOffsetInterval默认配置也为10秒
this.timer.scheduleAtFixedRate(new TimerTask() {
@Override
public void run() {
try {
// 持久化每个队列消费的偏移量
if (started.get()) ScheduleMessageService.this.persist();
} catch (Throwable e) {
log.error("scheduleAtFixedRate flush exception", e);
}
}
}, 10000, this.defaultMessageStore
.getMessageStoreConfig().getFlushDelayOffsetInterval());
}
}
start
方法执行之后,每个延迟级别都创建自己的定时任务,这里的定时任务的具体实现就在DeliverDelayedMessageTimerTask类之中,它核心代码是executeOnTimeup方法之中,我们来看一下主要部分:// 根据主题和队列Id获取消息队列
ConsumeQueue cq =
ScheduleMessageService.this.defaultMessageStore.findConsumeQueue(
TopicValidator.RMQ_SYS_SCHEDULE_TOPIC
, delayLevel2QueueId(delayLevel));
// 根据消费偏移量从消息队列中获取所有有效消息
SelectMappedBufferResult bufferCQ = cq.getIndexBuffer(this.offset);
// 遍历所有消息
for (; i < bufferCQ.getSize(); i += ConsumeQueue.CQ_STORE_UNIT_SIZE) {
// 获取消息的物理偏移量
long offsetPy = bufferCQ.getByteBuffer().getLong();
// 获取消息的物理长度
int sizePy = bufferCQ.getByteBuffer().getInt();
long tagsCode = bufferCQ.getByteBuffer().getLong();
// 省略部分代码...
long now = System.currentTimeMillis();
// 计算消息应该被消费的时间
long deliverTimestamp = this.correctDeliverTimestamp(now, tagsCode);
// 计算下一条消息的偏移量
nextOffset = offset + (i / ConsumeQueue.CQ_STORE_UNIT_SIZE)
long countdown = deliverTimestamp - now;
// 省略部分代码...
}
countdown
毫秒后再执行任务。如果到消费的时间,就继续执行下面操作:// 根据消息的物理偏移量和大小获取消息
MessageExt msgExt = ScheduleMessageService.this.defaultMessageStore.lookMessageByOffset(offsetPy, sizePy);
// 重新构建新的消息,包括:
// 1.清除消息的延迟级别
// 2.恢复真正的消息主题和队列Id
MessageExtBrokerInner msgInner = this.messageTimeup(msgExt);
// 重新把消息发送到真正的消息队列上
PutMessageResult putMessageResult =
ScheduleMessageService.this.writeMessageStore
.putMessage(msgInner);