ActiveMQ
from version 5.4 has an optional persistent scheduler built into the ActiveMQ message broker. It is enabled by setting the broker schedulerSupport attribute to true in the Xml Configuration.
<broker xmlns="http://activemq.apache.org/schema/core" brokerName="DorbaeBroker" useJmx="true" schedulerSupport="true">
.
.
.
</broker>
Schedule Properties
| Property name | Type | Description |
| AMQ_SCHEDULED_DELAY | long | The time in milliseconds that a message will wait before being scheduled to be delivered by the broker |
| AMQ_SCHEDULED_PERIOD | long | The time in milliseconds to wait after the start time to wait before scheduling the message again |
| AMQ_SCHEDULED_REPEAT | int | If set period, the number of times to repeat scheduling a message for delivery.
Total message count = 1 + repeat
If set cron, the number of message repeat when execute cron |
| AMQ_SCHEDULED_CRON | String | Use a Cron entry to set the schedule |
※ Be careful
If set AMQ_SCHEDULED_PERIOD, AMQ_SCHEDULED_REPEAT is the number of times to repeat scheduling a message for delivery.
(Total message count = 1 + repeat)
If set AMQ_SCHEDULED_CRON, AMQ_SCHEDULED_REPEAT is the number of times to repeat send message per 1 execution.
For example, you set
AMQ_SCHEDULED_REPEAT = 5
AMQ_SCHEDULED_CRON = * * * * *
6(1+5) messages are sent in destination per minute.
Client Sample
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
| package pe.dorbae.test.jms.activemq;
import java.util.Properties;
/*
* ************************************************************************************
*
* ScheduledMessage.java
*
* ************************************************************************************
*
* @version 1.0.00 2016-07-22 dorbae Initialize
* @since 2016-07-22
* @author dorbae
*
* ************************************************************************************
*/
public class TestScheduledMessage {
public TestScheduledMessage() {
// TODO Auto-generated constructor stub
}
public static void main(String[] args) {
String jmsFactory = "org.apache.activemq.jndi.ActiveMQInitialContextFactory";
String connectionFactoryName = "ConnectionFactory";
String jmsUrl = "tcp://localhost:61616";
String clientId= "iSharkClient.Test.ScheduledMessage";
String queueName = "DORBAE.TEST";
Properties jmsProps = new Properties();
jmsProps.put( javax.naming.Context.INITIAL_CONTEXT_FACTORY, jmsFactory);
jmsProps.put( javax.naming.Context.PROVIDER_URL, jmsUrl);
javax.naming.InitialContext jndi = null;
javax.jms.ConnectionFactory connectionFactory = null;
javax.jms.Connection conn = null;
javax.jms.Session session = null;
javax.jms.Destination destination = null;
javax.jms.Message message = null;
javax.jms.MessageProducer producer = null;
try {
jndi = new javax.naming.InitialContext( jmsProps);
connectionFactory = ( javax.jms.ConnectionFactory)jndi.lookup( connectionFactoryName);
conn = connectionFactory.createConnection();
conn.setClientID( clientId);
session = conn.createSession( false, javax.jms.Session.AUTO_ACKNOWLEDGE); // Not Transacted
destination = session.createQueue( queueName);
producer = session.createProducer( destination);
message = session.createTextMessage( "http://dorbae.blogspot.com");
/*
* For Delay Message
*
* Message will be sent to destination after delay times
*
*
long delayTime = 60L * 1000L; // 1 Minute
message.setLongProperty( org.apache.activemq.ScheduledMessage.AMQ_SCHEDULED_DELAY, delayTime);
*
*/
/*
* For Period Repeat
*
* Repeated messages will be sent to destination waiting period times between each re-delivery
*
long period = 10L * 1000L;
int repeat = 5;
message.setLongProperty( org.apache.activemq.ScheduledMessage.AMQ_SCHEDULED_PERIOD, period);
message.setIntProperty( org.apache.activemq.ScheduledMessage.AMQ_SCHEDULED_REPEAT, repeat);
*
*/
/*
* For Cron
*
* Message will be sent to destination according as cron pattern
*
*/
message.setStringProperty( org.apache.activemq.ScheduledMessage.AMQ_SCHEDULED_CRON, "0 * * * *");
producer.send( message);
} catch ( Exception e) {
e.printStackTrace();
} finally {
if ( session != null)
try {
session.close();
} catch ( Exception e) {}
if ( conn != null)
try {
conn.close();
} catch ( Exception e) {}
}
}
}
|
| ScheduledMessage.AMQ_SCHEDULED_CRON | AMQ_SCHEDULED_CRON |
| ScheduledMessage.AMQ_SCHEDULED_DELAY | AMQ_SCHEDULED_DELAY |
| ScheduledMessage.AMQ_SCHEDULED_ID | scheduledJobId |
| ScheduledMessage.AMQ_SCHEDULED_PERIOD | AMQ_SCHEDULED_PERIOD |
| ScheduledMessage.AMQ_SCHEDULED_REPEAT | AMQ_SCHEDULED_REPEAT |
| ScheduledMessage.AMQ_SCHEDULER_ACTION | AMQ_SCHEDULER_ACTION |
| ScheduledMessage.AMQ_SCHEDULER_ACTION_BROWSE | BROWSE |
| ScheduledMessage.AMQ_SCHEDULER_ACTION_END_TIME | ACTION_END_TIME |
| ScheduledMessage.AMQ_SCHEDULER_ACTION_REMOVE | REMOVE |
| ScheduledMessage.AMQ_SCHEDULER_ACTION_REMOVEALL | REMOVEALL |
| ScheduledMessage.AMQ_SCHEDULER_ACTION_START_TIME | ACTION_START_TIME |
| ScheduledMessage.AMQ_SCHEDULER_MANAGEMENT_DESTINATION | ActiveMQ.Scheduler.Management |
※ Be careful
The message property
scheduledJobId is reserved for use by the Job Scheduler.
If this property is set before sending, the message will be sent immediately and not scheduled. Also, after a scheduled message is received, the property scheduledJobId will be set on the received message so keep this in mind if using something like a Camel Route which might automatically copy properties over when re-sending a message.
A Scheduled message is canceled only using JMX unfortunately. There's an open issue to support rescheduling.
