ActiveMQ笔记(6)：消息延时投递

2022-05-16 21:41:18

在开发业务系统时，某些业务场景需要消息定时发送或延时发送（类似：飞信的短信定时发送需求），这时候就需要用到activemq的消息延时投递，详细的文档可参考官网说明，本文只介绍二种常用的用法：

注：本文采用spring的JmsTemplate来发送消息

步骤1、首先要修改activemq.xml配置文件，启用延时投递

 <broker xmlns="http://activemq.apache.org/schema/core" ... schedulerSupport="true" >

     ...

   </broker>

即：在broker节点加上schedulerSupport="true"，然后重启activemq即可

步骤2、定义一个MessagePostProcessor的实现类

import javax.jms.JMSException;

import javax.jms.Message;

import lombok.Data;

import org.apache.activemq.ScheduledMessage;

import org.apache.commons.lang3.StringUtils;

import org.springframework.jms.core.MessagePostProcessor;

/**

 * MQ延时投递处理器（注：ActiveMQ的配置文件中，要配置schedulerSupport="true"，否则不起作用）

 * by: 杨俊明 2016-06-16

 */

@Data

public class ScheduleMessagePostProcessor implements MessagePostProcessor {

    private long delay = 0l;

    private String corn = null;

    public ScheduleMessagePostProcessor(long delay) {

        this.delay = delay;

    }

    public ScheduleMessagePostProcessor(String cron) {

        this.corn = cron;

    }

    public Message postProcessMessage(Message message) throws JMSException {

        if (delay > 0) {

            message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, delay);

        }

        if (!StringUtils.isEmpty(corn)) {

            message.setStringProperty(ScheduledMessage.AMQ_SCHEDULED_CRON, corn);

        }

        return message;

    }

}

步骤3、jmsTemplate发送示例

        Object message1 = "corn消息内容：" + DateUtil.formatDate(new Date());

        //分 时 天 月 星期几

        jmsTemplate.convertAndSend(message1, new ScheduleMessagePostProcessor("40 22 * * *"));

        logger.info("消息1：[" + message1 + "] 延时发送成功！");

        jmsTemplate.convertAndSend(message1, new ScheduleMessagePostProcessor("50 22 * * *"));

        logger.info("消息1：[" + message1 + "] 延时发送成功！");

        Object message2 = "message：" + DateUtil.formatDate(new Date());

        jmsTemplate.convertAndSend(message2, new ScheduleMessagePostProcessor(30 * 1000));//延时30秒

        jmsTemplate.convertAndSend(message2, new ScheduleMessagePostProcessor(3600 * 24 * 1000));//延时24小时

        logger.info("消息2：[" + message2 + "] 延时发送成功！");

上面的代码演示了二种延时的用法：延时N毫秒、按corn表达式延时（注：此corn表达式并非Quartz框架中的corn表达式，而是linux中corntab中的表达式，基本顺序是"分(0-59) 时(0-23) 日(1-31) 月（1-12）星期几(1-7) "）

发送成功后，可以登录activemq的webconsole查看消息的属性：

在scheduled面板中，可以看到延时的消息

注：在开启消息持久化存储的前提下，就算把相应的queue在webconsole面板中删除（即删除队列），只要投递的时间尚未到，该消息也不会删除，仍然能正常延时投递。

此外，在queues面板中，如何查看某条具体的消息，也可以通过属性发现这条消息是延时消息，参考下图：

参考文章：
1、Delay and Schedule Message Delivery

2、喂鸡百科上的Corn表达式解释（中文）

3、喂鸡百科上的Corn表达式解释（英文）

4、kahaDB官方文档

码农公寓

相关文章