ActiveMQ使用手册(5)

((ActiveMQConnection)connection).setUseAsyncSend(true);

2.6.2 消费者异步分派

在ActiveMQ中，支持ActiveMQ以同步或异步模式向消费者分派消息。这样的意义：可以以异步模式向处理消息慢的消费者分配消息；以同步模式向处理消息快的消费者分配消息。

ActiveMQ默认以同步模式分派消息，这样的设置可以提高性能。但是对于处理消息慢的消费者，需要以异步模式分派。

在ConnectionFactory层面配置同步分派:

((ActiveMQConnectionFactory)connectionFactory).setDispatchAsync(false);

在Connection层面配置同步分派，此层面的设置将覆盖ConnectionFactory层面的设置： ((ActiveMQConnection)connection).setDispatchAsync(false);

在消费者层面以Destination URI配置同步分派，此层面的设置将覆盖ConnectionFactory和Connection层面的设置:

queue = new ActiveMQQueue(\

consumer = session.createConsumer(queue);

2.6.3 消费者优先级

在ActveMQ分布式环境中，在有消费者存在的情况下，如果更希望ActveMQ发送消息给消费者而不是其他的ActveMQ到ActveMQ的传送，可以如下设置： Java客户端：

queue = new ActiveMQQueue(\consumer = session.createConsumer(queue);

2.6.4 独占消费者

ActiveMQ维护队列消息的顺序并顺序把消息分派给消费者。但是如果建立了多个Session和MessageConsumer，那么同一时刻多个线程同时从一个队列中接收消息时就并不能保证处理时有序。

有时候有序处理消息是非常重要的。ActiveMQ支持独占的消费。ActiveMQ挑选一个MessageConsumer，并把一个队列中所有消息按顺序分派给它。如果消费者发生故障，那么ActiveMQ将自动故障转移并选择另一个消费者。可以如下设置： Java客户端：

queue = new ActiveMQQueue(\consumer = session.createConsumer(queue);

2.6.5 消息重发策略

在事务控制里抛出异常,txManager会进行rollback处理.(在activeMQ里,消息默认会redelivery到客户端6次,如果继续异常,消息会放到deadletter queue里(ActiveMQ.DLQ))。

在以下三种情况中，消息会被再次传送给消费者： 1．在使用事务的Session中，调用rollback()方法；

2．在使用事务的Session中，调用commit()方法之前就关闭了Session;

第18页

3．在Session中使用CLIENT_ACKNOWLEDGE签收模式，并且调用了recover()方法。

可以通过设置ActiveMQConnectionFactory和ActiveMQConnection来定制想要的再次传送策略。你可以在ActiveMQConnectionFactory 或ActiveMQConnection类中配置RedeliveryPolicy属性，用来定义重传策略的具体细节。

您可以使用Java代码，Spring配置或配置URI字符串来定义重传策略 可用的属性：

属性 collisionAvoidanceFactor 默认值 0.15 maximumRedeliveries 6 maximumRedeliveryDelay -1 initialRedeliveryDelay redeliveryDelay 1000L 1000L useCollisionAvoidance false useExponentialBackOff false backOffMultiplier 5 说明 设置防止冲突范围的正负百分比，只有启用useCollisionAvoidance参数时才生效。 最大重传次数，达到最大重连次数后抛出异常。为-1时不限制次数，为0时表示不进行重传。 最大传送延迟，只在useExponentialBackOff为true时有效（V5.5），假设首次重连间隔为10ms，倍数为2，那么第二次重连时间间隔为 20ms，第三次重连时间间隔为40ms，当重连时间间隔大的最大重连时间间隔时，以后每次重连时间间隔都为最大重连时间间隔。 初始重发延迟时间 重发延迟时间，当initialRedeliveryDelay=0时生效（v5.4） 启用防止冲突功能，因为消息接收时是可以使用多线程并发处理的，应该是为了重发的安全性，避开所有并发线程都在同一个时间点进行消息接收处理。所有线程在同一个时间点处理时会发生什么问题呢？应该没有问题，只是为了平衡broker处理性能，不会有时很忙，有时很空闲。 启用指数倍数递增的方式增加延迟时间。 重连时间间隔递增倍数，只有值大于1和启用useExponentialBackOff参数时才生效

2.6.6 目标相关的属性

ActiveMQ支持混合目标技术。它允许在一个JMS目标中使用一组JMS目标。

第19页

例如可以利用混合目标在同一操作中用向12个队列发送同一条消息或者在同一操作中向一个主题和一个队列发送同一条消息。

在混合目标中，通过“,”来分隔不同的目标。 Java客户端： 例如：

// send to 3 queues as one logical operation

Queue queue = new ActiveMQQueue(\producer.send(queue, someMessage);

如果在一个目标中混合不同类别的目标，可以通过使用“queue://”和“topic://”前缀来识别不同的目标。 例如：

// send to queues and topic one logical operation

Queue queue = new ActiveMQQueue(\

producer.send(queue, someMessage);

目标选项有：

属性 consumer.prefetchSize 默认值 variable 描述 The number of message the consumer will prefetch. Use to control if messages xare dropped if a slow consumer situation exists. Same as the noLocal flag on a Topic consumer. Exposed here so that it can be used with a queue. Should the broker dispatch messages asynchronously to the consumer. Is this a Retroactive Consumer. JMS Selector used with the consumer. consumer.maximumPendingMessageLimit 0 consumer.noLocal FALSE consumer.dispatchAsync FALSE consumer.retroactive FALSE consumer.selector null 第20页

consumer.exclusive FALSE consumer.priority Is this an Exclusive Consumer. Allows you to 0 configure a Consumer Priority. Java客户端的使用： 例如： queue = new

ActiveMQQueue(\er.prefetchSize=10\

consumer = session.createConsumer(queue);

2.6.7 消息预取处理

ActiveMQ的目标之一就是高性能的数据传送，所以ActiveMQ使用“预取限制”来控制有多少消息能及时的传送给任何地方的消费者。 一旦预取数量达到限制，那么就不会有消息被分派给这个消费者直到它发回签收消息（用来标识所有的消息已经被处理）。

可以为每个消费者指定消息预取。如果有大量的消息并且希望更高的性能，那么可以为这个消费者增大预取值。如果有少量的消息并且每条消息的处理都要花费很长的时间，那么可以设置预取值为1，这样同一时间，ActiveMQ只会为这个消费者分派一条消息。 Java客户端：

在ConnectionFactory层面为所有消费者配置预取值: tcp://localhost:61616?jms.prefetchPolicy.all=50 在ConnectionFactory层面为队列消费者配置预取值:

第21页

tcp://localhost:61616?jms.prefetchPolicy.queuePrefetch=1 使用“目标选项”为一个消费者配置预取值: queue = new

ActiveMQQueue(\

consumer = session.createConsumer(queue);

2.6.8 配置连接URL

ActiveMQ支持通过Configuration URI明确的配置连接属性。 例如：当要设置异步发送时，可以通过在Configuration URI中使用jms.$PROPERTY来设置。

tcp://localhost:61616?jms.useAsyncSend=true

以下的选项在URI必须以“jms.”为前缀。

属性 默认值 描述 If this flag is set then a seperate thread is not used for dispatching messages for each Session in the Connection. However, a separate thread is always used if there is more than one session, or the session isn't in auto acknowledge or dups ok mode Sets the JMS clientID to use null for the connection Sets the timeout before a close is considered complete. Normally a close() on a connection waits for confirmation from the broker; this allows that operation to timeout to save the client hanging if there is no broker. alwaysSessionAsync TRUE clientID closeTimeout 15000 (milliseconds) 第22页

百度搜索“77cn”或“免费范文网”即可找到本站免费阅读全部范文。收藏本站方便下次阅读，免费范文网，提供经典小说综合文库ActiveMQ使用手册(5)在线全文阅读。