11.4 Retroactive consumers

For applications that require messages to be sent and consumed as quickly as possible—for example, a real-time data feed—it’s recommend that you send messages withpersistence turned off.

11.4 消费者的消息追溯

对于要求消息能够尽快发送和接收处理的应用程序来说–例如一个实时的数据推送程序–推荐发送非持久化消息.

There’s a downside to consuming nonpersistent messages, in that you’ll only beable to consume messages from the point when your message consumer starts. Youcan miss messages if your message consumer starts behind the message producer, orthere’s a network glitch and your message consumer needs to reconnect to the broker(or another one in a network).

使用非持久化消息有一个弊端,即,你只能使用消息消费者启动之后的消息.如果消息消费者在消息生产者之后启动,你可能会丢失消息,或者,网络出现故障你的消息消费这需要重新连接代理(或代理网络中的其他代理)时也可能会丢失消息.

In order to provide a limited method of retroactive consumption of messages withoutrequiring message persistence, ActiveMQ has the ability to cache a configurablesize or number of messages sent on a topic. There are two parts to this—your messageconsumers need to inform the ActiveMQ broker that it’s interested in retroactive messages,and you need to configure the destination in the broker to say how many messagesshould be cached for consumption at a later point.

为了提供有限的追溯非持久化消息消费的方法,ActiveMQ为消息主题提供了可配置缓存大小的消息缓存.为此需要做两件事:你的消息消费者需要通知ActiveMQ代理,表示对可追溯消息感兴趣;并且,你需要配置代理中消息目的地以便代理能够决定为稍后的消息消费者缓存多少数量的消息.

To mark a consumer as being retroactive, you need to set the retroactive flag forthe message consumer. The easiest way to do that is to set the property on the topicname you use:

为了让消息消费者具有追溯消息能力,你需要为消息消费者设置retroactive开关.最简单的设置方式是在主题的属性中设置:

String brokerURI = ActiveMQConnectionFactory.DEFAULT_BROKER_URL;
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(brokerURI);
Connection connection = connectionFactory.createConnection();
connection.start();
 
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Topic topic = session.createTopic("soccer.division1.leeds?consumer.retroactive=true");
MessageConsumer consumer = session.createConsumer(topic);
Message result = consumer.receive();

On the broker side, you can configure a number of recovery policies on a topic-by-topicbasis. The default is called the FixedSizedSubscriptionRecoveryPolicy, whichholds a number of messages in a topic, based on the calculated size the messages willtake from the broker memory. The default size is 64 KB.

在代理端,你可以一个主题一个主题的配置一些恢复策略.默认的策略为FixedSizedSubscriptionRecoveryPolicy,该策略会保存一些主题中的消息,保存的消息数量根据即将从代理内存中取出的消息大小计算.消息大小的默认值为64KB.

You can configure the subscription recovery policy on a named topic, or use wildcardsto apply them to hierarchies. Here’s a sample configuration snippet of how tochange the default cache size for the FixedSizedSubscriptionRecoveryPolicy forall topics created in the ActiveMQ broker:

你可以为已命名的主题配置订阅的恢复策略,或者通过通配符将配置递归地应用到所有满足通配符的主题.下面是一个配置代码片段,该代码片段中使用新的配置为改变了ActiveMQ代理中FixedSizedSubscriptionRecoveryPolicy策略的默认缓存值.

<destinationPolicy>
<policyMap>
<policyEntries>
<policyEntry topic=">">
<subscriptionRecoveryPolicy>
<fixedSizedSubscriptionRecoveryPolicy maximumSize="8mb"/>
</subscriptionRecoveryPolicy>
</policyEntry>
</policyEntries>
</policyMap>
</destinationPolicy>

Retroactive message consumers are a convenient mechanism to improve the reliabilityof your applications without incurring the overhead of message persistence. We’veseen how enable retroactive consumers and how to configure their broker-side counterpart,the SubscriptionRecoveryPolicy.

消息消费者的消息追溯这种便利机制在不产生消息持久化开销的情况下改善了程序的可靠性.我们已经看到如何开启消息消费者的消息追溯以及如何配置代理端的SubscriptionRecoveryPolicy.

In the next section we’re going to look at how ActiveMQ stores messages that can’tbe delivered to message consumers—dead-letter queues.

下一节,我们将看到ActiveMQ如何存储无法分发到消息消费者的消息–死信队列.