ActiveMQ消息的延时和定时投递

- 日理万妓 2022-07-14 13:49 210阅读 0赞

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

<table style="font-family:verdana,arial,helvetica,sans-serif; line-height:24px; font-size:10px; color:rgb(0,0,0); margin:5px; border-collapse:collapse"> 
 <tbody> 
  <tr style="font-size:10px"> 
   <th style="border:1px solid rgb(204,204,204); padding:3px 4px; text-align:center; background-color:rgb(240,240,240)"> Property name</th> 
   <th style="border:1px solid rgb(204,204,204); padding:3px 4px; text-align:center; background-color:rgb(240,240,240)"> type</th> 
   <th style="border:1px solid rgb(204,204,204); padding:3px 4px; text-align:center; background-color:rgb(240,240,240)"> description</th> 
  </tr> 
  <tr style="font-size:10px"> 
   <td style="font-family:verdana,arial,helvetica,sans-serif; font-size:10px; border:1px solid rgb(204,204,204); padding:3px 4px"> AMQ_SCHEDULED_DELAY</td> 
   <td style="font-family:verdana,arial,helvetica,sans-serif; font-size:10px; border:1px solid rgb(204,204,204); padding:3px 4px"> long</td> 
   <td style="font-family:verdana,arial,helvetica,sans-serif; font-size:10px; border:1px solid rgb(204,204,204); padding:3px 4px"> 延迟投递的时间</td> 
  </tr> 
  <tr style="font-size:10px"> 
   <td style="font-family:verdana,arial,helvetica,sans-serif; font-size:10px; border:1px solid rgb(204,204,204); padding:3px 4px"> AMQ_SCHEDULED_PERIOD</td> 
   <td style="font-family:verdana,arial,helvetica,sans-serif; font-size:10px; border:1px solid rgb(204,204,204); padding:3px 4px"> long</td> 
   <td style="font-family:verdana,arial,helvetica,sans-serif; font-size:10px; border:1px solid rgb(204,204,204); padding:3px 4px"> 重复投递的时间间隔</td> 
  </tr> 
  <tr style="font-size:10px"> 
   <td style="font-family:verdana,arial,helvetica,sans-serif; font-size:10px; border:1px solid rgb(204,204,204); padding:3px 4px"> AMQ_SCHEDULED_REPEAT</td> 
   <td style="font-family:verdana,arial,helvetica,sans-serif; font-size:10px; border:1px solid rgb(204,204,204); padding:3px 4px"> int</td> 
   <td style="font-family:verdana,arial,helvetica,sans-serif; font-size:10px; border:1px solid rgb(204,204,204); padding:3px 4px"> 重复投递次数</td> 
  </tr> 
  <tr style="font-size:10px"> 
   <td style="font-family:verdana,arial,helvetica,sans-serif; font-size:10px; border:1px solid rgb(204,204,204); padding:3px 4px"> AMQ_SCHEDULED_CRON</td> 
   <td style="font-family:verdana,arial,helvetica,sans-serif; font-size:10px; border:1px solid rgb(204,204,204); padding:3px 4px"> String</td> 
   <td style="font-family:verdana,arial,helvetica,sans-serif; font-size:10px; border:1px solid rgb(204,204,204); padding:3px 4px"> Cron表达式</td> 
  </tr> 
 </tbody> 
</table>

下面我们演示一下间隔性重复投递；

生产者：

package cn.slimsmart.study.activemq;
    
    import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.DeliveryMode;
    import javax.jms.Destination;
    import javax.jms.MessageProducer;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    
    import org.apache.activemq.ActiveMQConnection;
    import org.apache.activemq.ActiveMQConnectionFactory;
    import org.apache.activemq.ScheduledMessage;
    
    public class Producer {
    
    	public static final String broker_url = "failover:(tcp://10.1.199.169:61616)";
    	private static String queue_name = "test.queue";
    
    	public static void main(String[] args) throws Exception {
    		ConnectionFactory factory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER, ActiveMQConnection.DEFAULT_PASSWORD, broker_url);
    		// 通过工厂创建一个连接
    		Connection connection = factory.createConnection();
    		// 启动连接
    		connection.start();
    		// 创建一个session会话 事务 自动ack
    		Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
    		// 创建一个消息队列
    		Destination destination = session.createQueue(queue_name);
    		// 创建生产者
    		MessageProducer producer = session.createProducer(destination);
    		// 消息持久化
    		producer.setDeliveryMode(DeliveryMode.PERSISTENT);
    		TextMessage message = session.createTextMessage("test delay message:" + System.currentTimeMillis());
    		long time = 60 * 1000;// 延时1min
    		long period = 10 * 1000;// 每个10s
    		int repeat = 6;// 6次
    		message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, time);
    		message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_PERIOD, period);
    		message.setIntProperty(ScheduledMessage.AMQ_SCHEDULED_REPEAT, repeat);
    		// 发送消息
    		producer.send(message);
    		session.commit();
    		producer.close();
    		session.close();
    		connection.close();
    	}
    }

消费者代码：

package cn.slimsmart.study.activemq;
    
    import java.util.concurrent.CountDownLatch;
    
    import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.Destination;
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageConsumer;
    import javax.jms.MessageListener;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    
    import org.apache.activemq.ActiveMQConnection;
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    public class Consumer {
    
    	public static final String broker_url = "failover:(tcp://10.1.199.169:61616)";
    	private static String queue_name = "test.queue";
    
    	public static void main(String[] args) throws Exception {
    		ConnectionFactory factory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER, ActiveMQConnection.DEFAULT_PASSWORD, broker_url);
    		// 通过工厂创建一个连接
    		Connection connection = factory.createConnection();
    		// 启动连接
    		connection.start();
    		// 创建一个session会话 事务 自动ack
    		Session session = connection.createSession(Boolean.TRUE, Session.CLIENT_ACKNOWLEDGE);
    		// 创建一个消息队列
    		Destination destination = session.createQueue(queue_name);
    		// 创建消费者
    		MessageConsumer consumer = session.createConsumer(destination);
    		consumer.setMessageListener(new MessageListener() {
    			@Override
    			public void onMessage(Message message) {
    				try {
    					System.out.println("receive message ：" + ((TextMessage) message).getText());
    					message.acknowledge();
    				} catch (JMSException e) {
    					e.printStackTrace();
    				}
    			}
    		});
    		new CountDownLatch(1).await();
    	}
    }

参考：

1.[延时和定时消息投递][Link 1]

2.[消息属性定义][Link 2]

[Link 1]: http://activemq.apache.org/delay-and-schedule-message-delivery.html
[Link 2]: http://activemq.apache.org/activemq-message-properties.html