第十一章 高级客户端选项

               

                第十一章 高级客户端选项

章节导读

 

•    如何使用专属消费者
•    消息组
•    了解对流对象以及二进制大对象的支持
•    失败重连
•     调度消息传递

 

11.1 专属消费者

  

     当消息从代理中分发时,遵循着先进先出原则.但是如果在一个队列中有超过一个消费者,你就不能保证这个原则了.这是因为你没有用了控制发送消息线程的调度--即使所有消费者使用同一个连接.理想情况下,只有一个消息消费者,为了支持容错机制,你需要其他的消费者在主消费者失败之后接管它的工作.ActiveMq支持多个消息消费者在队列上,但只有一个能从队列中接受消息.

        

       11.1.1 选择专属消费者

 

        对于那些认为消息顺序很重要的应用程序来说,你需要保证一个队列只有一个消息消费者.ActiveMQ提供一个客户端选项以便只让一个活动的消息消费者处理队列消息.ActiveMQ消息代理会为队列选择消息消费者来处理消息.让代理来选择消息消费者的好处是,如果当前的消息消费者停止了或者失效了,代理可以选择并激活其他的消费者如果消息队列中既有标准的消费者也有排他消息消费者,那么ActiveMQ消息代理将仍然将消息发送给其中的一个排他消息消费者.如果所有排他性的消费者都处于非激活状态,但同时还有标准的消息消费者存在,这是队列的消息处理模式将切换为常规方式--所有消息会以循环方式发送到所有保持活动的标注消息消费者.

    可以使用以下代码创建专属消费者

        

queue = new ActiveMQQueue("TEST.QUEUE?consumer.exclusive=true");
consumer = session.createConsumer(queue);

 

    11.1.2  使用专属消费者提供分布式锁

 

            通常,你需要从外部资源广播消息是为了更新数据库中的记录,或者在一个文件后面追加逗号分割的数据,或者发布实时的原始数据.你可能会希望创建一个系统备份,以便应用程序在读取或者广播变化数据失败后,系统备份可以接管并继续工作.通常情况下,你可以依靠给资源加锁(行锁或文件锁)以保证同一时间只有一个进程可以操作数据或者广播消息到一个主题.但是,当你不打算接受使用数据库带来的额外开销,或者你想要在多个机器上运行同一个程序(但不能使用分布锁),此时,你可以使用排他性消息消费者功能来创建一个分布式锁.图12.2中我们展示了一个应用程序,该程序作为客户端可以实时的读取数据,我们打算让这个程序实现失效转移.我们只打算让一个客户端连接到数据源并分发事件,但是当这个客户端失效后,我们需要其他的客户端可以接管.

   

 为了使用排他性的消息消费者来创建分布式锁,我们的消息生产者只注册一个消息队列.如果消息消费者从队列中接收了消息,该消费者即被激活,接下来即可订阅实时的数据源并将实时的数据转换成JMS消息.下面的是消息生产者初始化分布式锁的代码片段:

 

        

public void start() throws JMSException {
 this.connection = this.factory.createConnection();
 this.connection.start();
 this.session = this.connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
 Destination destination = this.session.createQueue(this.queueName+ "?consumer.exclusive=true");
 Message message = this.session.createMessage();
 MessageProducer producer = this.session.createProducer(destination);
 producer.send(message);
 MessageConsumer consumer = this.session.createConsumer(destination);
 consumer.setMessageListener(this);
}

 在这个例子中,我们总是发送消息到已知的消息队列以便启动消息消费--这个步骤可以在管理进程内部完成.请注意,我们使用了Session.CLIENT_ACKNOWLEDGE模式来消费消息.尽管我们打算将自己标识成一个排他性的消费者--因此占有锁--我们不想从公共的消息队列中移除消息.这种情况下,如果,当前消息生产者失败了,其他的排他性生产者会被激活.在这个例子中.我们将MessageListener实现成下面代码片段所示的样子.如果当前生产者未激活,我们可以调用示例方法startProducing().如果是在一个真实的程序中,这个方法应该要订阅一个实时的数据源并将实时数据转换成JMS消息.

public void onMessage(Message message) {
if (message != null && this.active==false) {
this.active=true;
startProducing();
}
}

 (译注:这个程序的应用场景: 利用排他性消息消费者,实现分布式锁.?当前的这个类首先是一个消息消费者(并且是排他性的消费者),从一个数据源接收消息.因为是排他性的,所以可以在多个机器上运行这个类,但同一时刻只有一个是处于激活状态的.处于激活状态时,当这个消费者接收消息时,调用onMessage方法,开始startProducing();如果当前的消费着失败了,则其他的消费者接替它继续工作,再次调用onMessage()方法,再次调用startProducing();.这样就相当于使用排他性消息消费者的特性,构造了一个可以分布式运行的producer.)

11.2 消息群组

           消息群组是排他性消费者概念的进一步完善.不同于将所有消息都发送给单一的消费者,消息可以被分组,同一组消息会发送到同一个消费者,并且消息生产者通过设置消息头中的JMSXGroupID值给消息分组.ActiveMQ代理确保JMSXGroupID值相同的消息会被发送到相同的消费者.

 

   如果代理之前根据JMSXGroupID而选中的消费者变得不可用了(比如网络故障),代理会选择另外一个代理来接收之前的分组消息.

   使用消息群组非常简单.消息群组的定义由用户决定并通过消息生产者完成-对于一个队列来说群组必须是唯一的.所有的路由都由ActiveMQ代理完成.

   为了创建消息群组,你选在消息生产者发送消息之前设置字符型属性JMSXGroupID设置属性值,如下代码所示:

  

Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Queue queue = session.createQueue("group.queue");
MessageProducer producer = session.createProducer(queue);
Message message = session.createTextMessage("<foo>test</foo>");
message.setStringProperty("JMSXGroupID", "TEST_GROUP_A");
producer.send(message);

 这个例子展示了消息生产者创建后,将一个文本消息设置成为TEST_GROUP_A消息组中一员.

消息群组使用常规的消息消费者,因此接收群组中的消息时无需完成任何额外的工作,消息生产者定义消息属于哪个群组,以及ActiveMQ代理选择消息消费者以便接收分组的消息后,所有使用消息分组工作就完成了.ActiveMQ代理会使用标准的消息头属性JMSXGroupSeq为群组中的每个消息设置一个序列号.对于新的群组来说该序列号从1开始.但是,从消息消费者的角度看,你不能假设消费者接受到的新群组中的第一个消息的序列号JMSXGroupSeq值是1.如果一个消息群组的消费者失效了,任何属于该群组的消息会发送到一个新的消费者.为了确定消息消费者开始从一个新的群组接收消息,或从一个之前没有接触过的群组接受消息,第一个发送的新消息消费者的消息会设置一个布尔型的属性JMSXGroupFirstForConsumer.你可以通过检查这个属性来确定当前消费者接受到的消息是否是群组中的第一个消息,示例代码如下:

Session session = MessageConsumer consumer = session.createConsumer(queue);
Message message = consumer.receive();
String groupId = message.getStringProperty("JMSXGroupId");
if (message.getBooleanProperty("JMSXGroupFirstForConsumer")) {
// do processing for new group
}

 通常情况下,你会同时启动很多消息消费者处理消息.ActiveMQ会将所有消息群组平均分配给所有的消息消费者,但是如果消息群组中有正等待被分发的消息,那么这个群组通常被分配给第一个消息消费者.

为了更加均衡的分布负载,可以给代理一个信号让其等待更多的消息消费者启动.为此,你需要在ActiveMQ代理的XML配置文件中设置一个消息目的地策略.设置consumersBeforeDispatchStarts属性为你的应用在启动前最小需要的消息消费者的数量如下面示例代码所示:

<destinationPolicy>
<policyMap>
<policyEntries>
<policyEntry queue=">"   consumersBeforeDispatchStarts="2"  timeBeforeDispatchStarts="5000"/>
</policyEntries>
</policyMap>
</destinationPolicy>

 上面的示例配置代码通知ActiveMQ代理所有的队列(通过使用通配符>,匹配所有队列)都要等待两个消息消费者都启动完成了才开始分发消息.此外,我们还设置了timeBeforeDispatchStarts属性为5000毫秒,以便通知ActiveMQ代理如果两个消息消费者没有在5秒内成功接收到队列中的第一个消息,队列应该使用第一个可用的消费者.

   使用消费群组确实是会给ActiveMQ代理增加一些额外开销的,因为需要为每一个消息群组存储路由信息,可以通过设置JMSXGroupID为你打算关闭的群组名称同时设置JMSXGroupSeq值为-1,来明确的关闭一个消息群组.如下面的示例代码所示:

   

Session session =
connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Queue queue = session.createQueue("group.queue");
MessageProducer producer = session.createProducer(queue);<foo />
Message message = session.createTextMessage("<foo>close</foo>");
message.setStringProperty("JMSXGroupID", "TEST_GROUP_A");
message.setIntProperty("JMSXGroupSeq", -1);
producer.send(message);

 你可以通过发送新消息到已关闭的群组来重新创建这个群组.但是ActiveMQ代理可能会给这个重新创建的消息群组设置一个新的消息消费者.从概念上讲,消息群组类似与使用消息选择器,不同的是使用消息分组可以自动的为消息消费者选择消息,同时当群组的消费者失效后还可以进行失效转移.

11.3   使用ActiveMQ流

   ActiveMQ流是一个高级特性,允许你在ActiveMQ客户端使用Java的IOStream.ActiveMQ会将一个OutputStream拆分成不同的数据块,然后每一个数据库当成JMS消息并发送.在消息消费者端需要使用一个相对应的ActiveMQ JMS InputStream来重新组装接受到的数据块.

 如果使用消息队列作为流的消息目的地,使用多个消息消费者(或者使用排他性消息消费者)比较好,因为这个特性中用到了消息群组.这样会导致具有相同群组ID的消息会被发送到同一个消息消费者.这种场景下使用多个消息生产者会因为消息顺序问题产生问题.

   使用JMS流的好处是ActiveMQ将一个流分割成可管理的数据块并且可以在消息消费者中重新组装这些数据块.因此,使用ActiveMQ流可以传输非常大的文件,如图所示

 

 

FileInputStream in = new FileInputStream("largetextfile.txt");
String brokerURI = ActiveMQConnectionFactory.DEFAULT_BROKER_URL;
ConnectionFactory connectionFactory =  new ActiveMQConnectionFactory(brokerURI);
Connection connection = (ActiveMQConnection)
connectionFactory.createConnection();
connection.start();
Session session =
connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Queue destination = session.createQueue(QUEUE_NAME);
OutputStream out = connection.createOutputStream(destination);
//now write the file on to ActiveMQ
byte[] buffer = new byte[1024];
while(true){
int bytesRead = in.read(buffer);
if (bytesRead==-1){
break;
}
out.write(buffer,0,bytesRead);
}
//close the stream so the receiving side knows the steam is finished
out.close();

 我们使用connection创建了一个OutputStream.我们使用FileInputStream读取了一个文件,然后将FileInputStream写入到ActiveMQ的OutputStream中.

 

   需要注意的是,我们在完成了读取文件后关闭了ActiveMQ的OutoutStream.这点很重要,因为关闭了OutputStream后,ActiveMQ流的接收端就可以决定当前接受的流是否结束.推荐位每个要发送的文件使用新的OutputStream。

   为了完整起见,下面列出接收端的代码

   

FileOutputStream out = new FileOutputStream("copied.txt");
String brokerURI = ActiveMQConnectionFactory.DEFAULT_BROKER_URL;
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(brokerURI);
Connection connection = (ActiveMQConnection) connectionFactory.createConnection();
connection.start();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//we want to be an exclusive consumer
String exclusiveQueueName= QUEUE_NAME + "?consumer.exclusive=true";
Queue destination = session.createQueue(exclusiveQueueName);
InputStream in = connection.createInputStream(destination);
//now write the file from ActiveMQ
byte[] buffer = new byte[1024];
while(true){
int bytesRead = in.read(buffer);
if (bytesRead==-1){
break;
}
out.write(buffer,0,bytesRead);
}
out.close();

 你也可以在Topic中使用流,但如果消息消费者在流已经开始传送之后才启动,则该消费者不会接受到任何在它之前就已经发送过的任何数据.

 

ActiveMQ将流分成可管理的数据块然后将所有的数据块作为独立的消息发送.这就意味着你在使用流时必须十分小心.因为如果消息消费者在读取InputStream的中途失效了,那么已经被读取的数据将无法重现了.

  ActiveMQ流对于传输大数据来说十分有用,尽管你需要仔细考虑,以应对使用ActiveMQ流的应用程序的失效场景.对于发送大数据还有一种更健壮的方法:Blob消息。

  11.4   使用二进制消息

     ActiveMQ 引入了二进制消息的概念这样用户可以将消息分发的语义(传输连接,负载均衡和智能路由)同超大尺寸消息结合起来.二进制消息并不包含要发送的数据,而是通知要发送的二进制数据(大尺寸二进制对象)已经准备完成了.二进制对象本身是在消息之外传输的,通过FTP或者HTTP传输.事实上,Activemq的二进制消息仅包含二进制数据的URL,通过一个助手方法可以抓取InputStream进而获取正真的二进制数据.

   首先,我们来看看如何创建一个二进制消息.在下面这个例子中,我们假设在一个共享的站点上存在一个大尺寸的文件.因此,我们需要创建一个二进制消息通知所有感兴趣的消费者文件已经存在

import org.apache.activemq.BlobMessage;
...
String brokerURI = ActiveMQConnectionFactory.DEFAULT_BROKER_URL;
ConnectionFactory connectionFactory =
new ActiveMQConnectionFactory(brokerURI);
Connection connection = connectionFactory.createConnection();
connection.start();
ActiveMQSession session = (ActiveMQSession)
connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Queue destination = session.createQueue(QUEUE_NAME);
MessageProducer producer = session.createProducer(destination);
BlobMessage message =
session.createBlobMessage(new URL("http://example.com/bigfile.dat"));
producer.send(message);

    上门的示例代码中,我们创建了一个JMS连接,并且ActiveMQ的session的一个方法支持创建二进制消息.我们使用共享站点上一个文件的URL创建了一个二进制消息,并将这个二进制消息发送到已知的队列上去.

    下面是相应的消息消费者接受二进制消息的代码:

import org.apache.activemq.BlobMessage;
...
// destination of our Blob data
FileOutputStream out = new FileOutputStream("blob.dat");
String brokerURI = ActiveMQConnectionFactory.DEFAULT_BROKER_URL;
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(brokerURI);
Connection connection = (ActiveMQConnection)connectionFactory.createConnection();
connection.start();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Queue destination = session.createQueue(QUEUE_NAME);
MessageConsumer consumer = session.createConsumer(destination);
BlobMessage blobMessage = (BlobMessage) consumer.receive();
InputStream in = blobMessage.getInputStream();
// now write the file from ActiveMQ
byte[] buffer = new byte[1024];
while (true) {
int bytesRead = in.read(buffer);
if (bytesRead == -1) {
break;
}
out.write(buffer, 0, bytesRead);
}
out.close();

 使用二进制消息比使用流消息更加健壮,尽管两者都是自动工作的,但是两者都依赖外部服务器来存储实际要发送的数据.

11.5   使用失效转移协议应对代理或网络故障

    在前面的章节,介绍过失效转移协议,解释了在失效的情况下,允许客户端转移到另外一个ActiveMQ代理的知识.失效转移协议是ActiveMQ代理的Java客户端使用的默认协议,因此我们有必要了解更多协议相关的特性和功能.

     默认情况下,你给客户端的URL指定了一个或者多个ActiveMQ代理以便来创建连接:

failover:(tcp://host1:61616,tcp://host2:61616,ssl://host2:61616)

  这种方式指定URL是没有问题的,但是假如有一些嵌入的查询参数的话这种方式指定的URI看起来有些混乱.

  ActiveMQ将在指定的URI列表中随机悬着一个然后创建到该代理的连接.通过使用失效转移协议,ActiveMQ客户端会定期启动一个保持连接的协议,以便侦测代理是否不可达(连接丢失或代理丢失).如侦测到代理不再可用,失效转移连接器会从提供的URI列表中随机的选择另外一个代理的URI.如果仅提供了一个代理URI.客户端会隔一段时间后再检测这个唯一代理是否可用了.可以通过给ActiveMQ连接设置TransportListener来监听传输是否中断了,代码如下所示:

 

ActiveMQConnection connection = (ActiveMQConnection)
connectionFactory.createConnection();
connection.addTransportListener(new DefaultTransportListener() {
 public void onException(IOException arg0) {
 System.err.println("This is bad");
 }
 public void transportInterrupted() {
 System.out.println("Transport interrupted");
 }
 public void transportResumed() {
 System.out.println("Transport resumed"); 
 }
});
connection.start();

 如果你为失效转移连接器提供了多个URI,那么默认情况下,连接器会随机使用其中一个URI,以便保证客户端访问代理的负载均衡.当你需要确保客户端按照设定的顺序来连接指定的代理时,要设置failover:(tcp://master:61616,tcp://slave:61616)?randomize=false

 如果失效转移连接器中的URI都不可用,那么默认情况下,它会等待一段时间然后重现连接,相邻两次失败后等待时间会递增(这被称为指数级back-off).失效转移连接器默认会开启useExponentialBackOff.在重新尝试连接之前的等待时间称为initialReconnectDelay(初始值为10毫秒)下一次失败后尝试重连之前的等待时间与前一次等待时间的倍数称为backOffMultiplier(默认2.0),你也可以使用maxReconnectDelay(默认值为30000毫秒)设置等待间隔时间的最大值.

failover:(tcp://master:61616,tcp://slave:61616)?backOffMultiplier=1.5,initialReconnectDelay=1000

   使用基于TCP的传输连接器可能导致一个潜在的问题,即,连接的对端(对于ActiveMQ来说,这个对端指代理)是否已经失效了.导致这个问题的原因有几个,比如ActiveMQ代理失效或者网络连接丢失.

   另外,如果在ActiveMQ代理和客户端之间存在防火墙,如果连接长时间处于非活动状态,防火墙可能会切断连接.可以配置TCP连接的keepalive属性,但是这个属性是操作系统属性,可能会导致修改系统内容参数并且在复杂环境网络中不能很好的工作.鉴于此,ActiveMQ在传输连接器的上层使用一个保持连接协议以保持防火墙始终打开同时检测代理是否不可达.keepalive协议每隔一段时间会发送一个轻量级的命令消息给代理然后等待代理的响应.

  如果没有在规定时间内收到响应,ActiveMQ会认为当前的传输连接器不再可用.失效转移连接器监听到失效的连接器后会选择另外一个连接器来应对前面连接器的失效.Activemqs在keepalive协议中使用maxInactivityDuration参数是一个OpenWire属性,默认值是30000毫秒.你可以为失效转移协议的这个参数设置新值.如下代码所示 

 

failover:(tcp://host1:61616?wireFormat.maxInactivityDuration=1000,tcp://host2:61616?wireFormat.maxInactivityDuration=1000)

 你必须为使用失效转移协议的连接器设置这个参数(以及其它所有的OpenWire属性),而不是为失效转移本身设置这个.

 

    默认情况下,ActiveMQ客户端在发送消息时使用的分发模式是持久化的.使用持久化分发模式的消息将使用同步方式发送,这种方式发送消息后,发送消息的线程将会阻塞知道收到代理的回执以确定消息成功发送以及成功存储了.对于关注性能的程序来说,使用非持久化分发方式能显著提高性能,使用非持久化分发时,消息都是异步发送,这样就产生了隐患-如果传输连接器失效你可能会丢失消息.你可以通过配置失效转移连接器的TrackMessages属性来启用缓存以避免消息丢失.这种缓存消息的最大值也可以使用失效转移连接器的maxCacheSize配置-该配置的默认值是128KB(允许分配给消息缓存的内存大小).

  下面是示例配置代码,配置中通过设置最大缓存大小启用了消息缓存:

 

failover:(tcp://host1:61616,tcp://host2:61616)?trackMessages=true,maxCacheSize=256000

 对于搞性能程序来说,失效转移协议也是非常重要的,新建一个传输连接需要的时间是很可观的(几十到几百毫秒),因此启用了快速失效转移后,ActiveMQ能有选择性的运行失效转移协议创建一个备份连接,以便能在主链接失效时迅速的接替主链接继续工作.设置运行失效转移连接器建立备份连接的属性被称为backup.你可以通过backupPoolSize属性设置多于一个的备份(默认1)：

 

 

failover:(tcp://host1:61616,tcp://host2:61616,tcp://host3:61616)?backup=true,backupPoolSize=2

 目前为止,我们都是使用一个静态的URI列表来配置失效转移连接器的,但是ActiveMQ是可以自行监测需要连接的代理,因此ActiveMQ客户端可以在代理新增到集群或者从集群中移除时动态的更新代理集群,我们需要启用传输连接器的updateClusterClients属性来为ActiveMQ客户端启用动态更新代理集群.TransportConnector中用于控制代理集群自动更新的属性如表所示:

 

属性名	默认值	描述
updateClusterClients	false	如果设置true,当代理集群发生变化时会传递消息给已连接到集群的客户端.
rebalanceClusterClients	false	如果设置为true,当新代理加入集群时,客户端会被要求重新做负载均衡
updateClusterClientsOnRemove	false	如果设置为true,当代理集群从代理网络中移除时更新客户端,将这个属性独立出来是为了能够在新代理加入代理网络时通知客户端,但是移除代理时不通知
updateClusterFilter	null	逗号分隔的正则表达式列表,用于匹配代理名称,只有匹配上的代理加入到客户端失效转移代理集群中,才会更新客户端

rebalanceClusterClients是一个有趣的属性,启用该属性时,当一个新代理加入集群,ActiveMQ客户端会平均分配它们的负载到集群中所有的代理.

使用上述属性,在一台机器名为tokyo机器上配置ActiveMQ的示例代码如下:

 

<broker>
...
<transportConnectors>
<transportConnector name="clustered"
uri="tcp://0.0.0.0:61616"
updateClusterClients="true"
updateClusterFilter="*newyork*,*london*" />
</<transportConnectors>
...
</broker>

  上述配置将在代理名称中含有newyork或london的代理加入代理集群中时更新所有使用失效转移连接器的客户端.开启updateClusterClients后,你仅需要为集群中的一个代理配置失效转移协议,例如:

 

 

failover:(tcp://tokyo:61616)

 新的代理加入到集群中时客户端会自动更新,如果机器tokyo宕机了,客户端会自动的进行失效转移从而连接到机器名为newyork或london的代理.

 

你可能希望你的客户端能自动的针对代理集群中所有的集群进行分布式部署,这样所有的机器可以为消息程序均衡负载,通过启用传输连接器的rebalanceClusterClients属性,在ActiveMQ代理加入集群或从集群中移除代理时,上述客户端针对代理集群进行分布式部署将自动完成.

 

11.6   使用ActiveMQ消息调度延迟发送消息

    activeMQ消息调度,实现的消息延迟发送或者在按照固定时间的间隔实现间隔发送的功能十分有用,其中一个独一无二的好处是消息调度设置为延迟发送的消息将会被持久化存储,因而在ActiveMQ失效后不会丢失重启后会继续发送.你可以通过严格定义消息的属性来设置如何延迟发送消息.为方便起见,常用的延迟发送消息相关的属性都在org.apache.activemq.ScheduledMessage接口中有定义.如下表所示

属性	默认	描述
AMQ_SCHEDULED_DELAY	false	消息延迟发送的时间（毫秒）
AMQ_SCHEDULED_PERIOD	false	代理启动后,发送消息之前的等待时间
AMQ_SCHEDULED_REPEAT	false	调度消息发送的重复次数
AMQ_SCHEDULED_CRON	String	使用一个Cron实体设置消息发送调度

只需设置AMQ_SCHEDULED_DELAY这个属性即可让消息等待一段时间后再发送,假设你打算从你的客户端发送消息,但需要设置消息延迟五分钟后发送,你可以使用下面的代码实现

 

MessageProducer producer = session.createProducer(destination);
TextMessage message = session.createTextMessage("test msg");
long delayTime = 5 * 60 * 1000;
message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, delayTime);
producer.send(message);

 设置延迟发送之后,ActiveMQ将消息存储在代理中,等待设置的延时时间过了之后,消息会被发送到设定的目的地.尽管你已经指定了消息在5分钟之后发送,但是如果消息目的地是一个消息队列,则消息会被发送到队列的末端,这一点很重要.因而,消息发送的实际延迟时间将取决于当前消息的目的地队列中有多少个正等待的消息.

 

  你也可以使用AMQ_SCHEDULED_PERIOD和AMQ_SCHEDULED_REPEAT属性设置消息按照固定间隔时间发送固定的次数.下面的示例代码将发送消息100次,每次发送间隔时间为30秒:

MessageProducer producer = session.createProducer(destination);
TextMessage message = session.createTextMessage("test msg");
long delay = 30 * 1000;
long period = 30 * 1000;
int repeat = 99;
message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, delay);
message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_PERIOD, period);
message.setIntProperty(ScheduledMessage.AMQ_SCHEDULED_REPEAT,repeat);
producer.send(message);

 注意,我们将消息重复发送的次数设置为99,因为第一次发送消息 + 99此重复=100.如果你设置消息只被发送一次,那么消息的ID即使你设置的消息ID.如果你设置了重复发送,或者使用AMQ_SCHEDULED_CRON属性来调度消息发送,那么ActiveMQ会生产一个新的唯一的消息ID作为重复发送的消息ID.

 

 Cron是Unix系统中任务调度器,它使用一个字符串来表示一个任务何时需要被执行.ActiveMQ使用同样的预防,如下文本描述:

.---------------- minute (0 - 59)
| .------------- hour (0 - 23)
| | .---------- day of month (1 - 31)
| | | .------- month (1 - 12) - 1 = January
| | | | .---- day of week (0 - 7) (Sunday=0 or 7
| | | | |
* * * * *

 例如,如果你打算在每月的12号早上2点发送消息,你需要按照如下代码所示进行设置:

 

 

MessageProducer producer = session.createProducer(destination);
TextMessage message = session.createTextMessage("test msg");
message.setStringProperty(ScheduledMessage.AMQ_SCHEDULED_CRON,"0 2 12 * *");
producer.send(message);

 你可以同时使用cron和普通的延迟与重复来调度消息发送,但是cron方式的调度具有优先权.例如,不同于每月的12号早上2点只发送一条消息,你可能打算在每月的12号早上2点开始发送消息,并且每隔30秒再重复发送9个消息:

 

 

long delay = 30 * 1000;
long period = 30 * 1000;
int repeat = 9;
MessageProducer producer = session.createProducer(destination);
TextMessage message = session.createTextMessage("test msg");
message.setStringProperty(ScheduledMessage.AMQ_SCHEDULED_CRON,"0 2 12 * *");
message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, delay);
message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_PERIOD, period);
message.setIntProperty(ScheduledMessage.AMQ_SCHEDULED_REPEAT,repeat);
producer.send(message);