ActiveMQ消息的延时和定时投递

Dear 丶 2023-05-21 08:54 101阅读 0赞

## 基本介绍 ##

ActiveMQ对消息延时和定时投递做了很好的支持，其内部启动Scheduled来对该功能支持，也提供了一个封装的消息类型：`org.apache.activemq.ScheduledMessage`，只需要把几个描述消息定时调度方式的参数作为属性添加到消息，broker端的调度器就会按照我们想要的行为去处理消息。

各个参数如下表格：

<table> 
 <thead> 
  <tr> 
   <th>Property name</th> 
   <th>type</th> 
   <th>description</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>AMQ_SCHEDULED_DELAY</td> 
   <td>long</td> 
   <td>延迟投递的时间</td> 
  </tr> 
  <tr> 
   <td>AMQ_SCHEDULED_PERIOD</td> 
   <td>long</td> 
   <td>重复投递的时间间隔</td> 
  </tr> 
  <tr> 
   <td>AMQ_SCHEDULED_REPEAT</td> 
   <td>int</td> 
   <td>重复投递次数</td> 
  </tr> 
  <tr> 
   <td>AMQ_SCHEDULED_CRON</td> 
   <td>String</td> 
   <td>Cron表达式</td> 
  </tr> 
 </tbody> 
</table>

## 案例演示 ##

1.  需要在activemq.xml 中配置 schedulerSupport属性改为true（修改完成后重启activemq）  
    ![在这里插入图片描述][20200424152809408.png]
2.  在java代码中给message设置属性 使用`ScheduledMessage`类型来设置  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Z5ajEzOTI1NDc1OTU3_size_16_color_FFFFFF_t_70]

代码示例如下：

public class ActivemqProducer { 
        static String BROCKER_URL = "tcp://192.168.72.129:61616";
        static String QUEUE_NAME = "queue_name8";
    
        public static void main(String[] args) throws JMSException { 
            // 1. 创建连接工厂
            ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(BROCKER_URL);
            // 2. 获取连接 并启动
            Connection connection = activeMQConnectionFactory.createConnection();
            connection.start();
            // 3. 创建session ,参数一：事务，参数二：签收
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            //4. 创建目的地,队列
            Queue queue = session.createQueue(QUEUE_NAME);
            // 5.创建消息生产者 类型为ActiveMQMessageProducer
            MessageProducer messageProducer = session.createProducer(queue);
            messageProducer.setDeliveryMode(DeliveryMode.PERSISTENT);
    
            // 6.通过 messageProducer 发送3条消息
            for (int i = 1; i <= 3; i++) { 
                TextMessage textMessage = session.createTextMessage("消息 ->" + i);
                // 设置延迟投递的时间为3秒
                textMessage.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, 10*1000);
                // 设置重复投递的次数为3次
                textMessage.setIntProperty(ScheduledMessage.AMQ_SCHEDULED_REPEAT, 3);
                // 设置重复投递的间隔时间为2秒
                textMessage.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_PERIOD, 2*1000);
    
                messageProducer.send(textMessage);
            }
    
            // 7.关闭资源
            messageProducer.close();
            session.close();
            connection.close();
        }
    }

[20200424152809408.png]: /images/20230521/9f5479dbb8774d4c9aefda41d7477d1b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Z5ajEzOTI1NDc1OTU3_size_16_color_FFFFFF_t_70]: /images/20230521/ede29596f7454399897126e65b03f74c.png