直观的结果:当生产者向 topic 发送消息,

1. 若不存在持久订阅者和在线的普通订阅者,这个消息不会保存,当普通订阅者上线后,它是收不到消息的。

2. 若存在离线的持久订阅者,broker 会为该持久订阅者保存消息,当该持久订阅者上线后,会收到消息。

本质:producer 发送消息给 topic,broker 接收消息并且分发给 consumers。consumers 包括持久订阅者和在线的普通订阅者,对于持久订阅者,broker 把消息添加到它的 message cursor 中;对于普通订阅者,broker 直接分发消息。

如果,1个 topic 有2个持久订阅者,并且这2个持久订阅者都不在线,这时 producer 向 topic 发送1条消息,则 broker 会保存2条消息。因此,如果1个 topic 有很多不在线的持久订阅者,会导致 broker 消耗过多存储。

对于持久化的消息,这很好验证:为方便查看消息,将 broker 持久化方式配置为jdbc,则可以在 ACTIVEMQ_MSGS 表中看到持久化消息。

持久订阅者1:

new ActiveMQConnectionFactory("tcp://localhost:61616?jms.clientID=10086");
...
TopicSubscriber consumer = session.createDurableSubscriber(destination, "subscriber_zhang");

持久订阅者2:

new ActiveMQConnectionFactory("tcp://localhost:61616?jms.clientID=10087");
...
TopicSubscriber consumer = session.createDurableSubscriber(destination, "subscriber_zhang");

对于持久消息,验证 broker 为每个持久订阅者保存1条消息:
1. 启动 broker;
2. 分别启动2个持久订阅者,然后关闭它们,这样 broker 有了2个离线的持久订阅者;
3. 启动 producer 向 topic 发送1条消息;
4. 查看 ACTIVEMQ_MSGS 表

对于非持久消息,直接跟代码了,这里不说明。

下图是 Topic 接收消息并分发的调用栈:

ActiveMQ topic 普通订阅和持久订阅-LMLPHP

// org.apache.activemq.broker.region.policy.SimpleDispatchPolicy
// 分发策略很简单,就是遍历consumers
public boolean dispatch(MessageReference node, MessageEvaluationContext msgContext, List<Subscription> consumers)
throws Exception {
int count = 0;
for (Subscription sub : consumers) {
// Don't deliver to browsers
if (sub.getConsumerInfo().isBrowser()) {
continue;
}
// Only dispatch to interested subscriptions
if (!sub.matches(node, msgContext)) {
sub.unmatched(node);
continue;
} //持久化订阅是 DurableTopicSubscription
//普通订阅是 TopicSubscription
sub.add(node);
count++;
} return count > 0;
}

持久化订阅:

// org.apache.activemq.broker.region.DurableTopicSubscription
public void add(MessageReference node) throws Exception {
if (!active.get() && !keepDurableSubsActive) {
return;
}
// 调用 PrefetchSubscription.add
super.add(node);
} // org.apache.activemq.broker.region.PrefetchSubscription
public void add(MessageReference node) throws Exception {
synchronized (pendingLock) {
// The destination may have just been removed...
if( !destinations.contains(node.getRegionDestination()) && node!=QueueMessageReference.NULL_MESSAGE) {
// perhaps we should inform the caller that we are no longer valid to dispatch to?
return;
} // Don't increment for the pullTimeout control message.
if (!node.equals(QueueMessageReference.NULL_MESSAGE)) {
enqueueCounter++;
}
//首先加入到 message cursor 中,pending 类型为 StoreDurableSubscriberCursor
pending.addMessageLast(node);
}
dispatchPending();
}

持久化订阅者上线后,也会触发消息分发动作即 dispatchPending,调用栈如下图:

ActiveMQ topic 普通订阅和持久订阅-LMLPHP

持久化订阅者使用的 message cursor 是 StoreDurableSubscriberCursor。

普通订阅:

org.apache.activemq.broker.region.TopicSubscription.add(MessageReference node)

05-11 22:38