提问人:sbi 提问时间:11/10/2016 最后编辑:sbi 更新时间:11/22/2016 访问量:1868
如何使我的 ActiveMQ 代理删除脱机持久订阅者
How do I make my ActiveMQ broker drop offline durable subscribers
问:
我们有一个 ActiveMQ 代理,它使用 JMS、AMQP 和 MQTT 从非常不同的客户端连接到。出于某种原因,我们还没有弄清楚一组特定的 MQTT 客户端通常(并非总是)持久订阅。这是一个测试环境,其中经常添加和删除客户端,后者有时通过拔掉插头或重新启动嵌入式设备来添加和删除客户端,因此它们无法正确取消订阅。其效果 (IIUC) 是代理为它可能再也看不到的设备堆积了“离线持久订阅”(我可以在 http://my_broker:8161/admin/subscribers.jsp 下看到这些),永远保留有关这些主题的消息,直到它最终在自己的内存占用下崩溃。
这里的问题是订阅者持久订阅,我们需要找出为什么会这样。然而,我们也决定,这样做的客户(在不知不觉中)不应该让经纪人停滞不前,所以我们需要独立解决这个问题。
我发现有离线持久订阅的超时设置,并将其放入我们的代理配置中(最后两行):
<broker
xmlns="http://activemq.apache.org/schema/core"
brokerName="my_broker"
dataDirectory="${activemq.data}"
useJmx="true"
advisorySupport="false"
persistent="false"
offlineDurableSubscriberTimeout="1800000"
offlineDurableSubscriberTaskSchedule="60000">
如果我理解正确的话,上述内容应该每分钟检查一次,并解雇半小时未见的客户。然而,与文档相反,这似乎行不通:我订阅了几天后拔掉插头的消费者在离线持久订阅者列表中仍然可见,代理的内存占用量不断增加,如果我在代理的 Web 界面中手动删除订阅者,我可以看到内存占用量在下降。
所以这是我的问题:
- 什么决定了对 ActiveMQ 代理上主题的 MQTT 订阅是否持久?
- 在 ActiveMQ 设置中设置删除离线持久订阅的超时时,我做错了什么?
答:
我提取了删除超时持久订阅的相关代码 ()。doCleanup()
在成功的情况下,它会执行:
LOG.info("Destroying durable subscriber due to inactivity: {}", sub);
在失败的情况下,它会执行:
LOG.error("Failed to remove inactive durable subscriber", e);
在日志文件中查找上述日志行,并将其与您使用 admin/subscribers.jsp 查看器观察到的详细信息进行匹配。如果它没有打印任何行,则订阅可能由于某种原因而保留,或者您可能偶然发现了错误。active
另外,如果可以的话,您可以尝试删除经纪人名称中的下划线 (_) 吗?该手册讨论了代理名称中下划线的问题。
法典:
public TopicRegion(RegionBroker broker, DestinationStatistics destinationStatistics, SystemUsage memoryManager, TaskRunnerFactory taskRunnerFactory, DestinationFactory destinationFactory) {
super(broker, destinationStatistics, memoryManager, taskRunnerFactory, destinationFactory);
if (broker.getBrokerService().getOfflineDurableSubscriberTaskSchedule() != -1 && broker.getBrokerService().getOfflineDurableSubscriberTimeout() != -1) {
this.cleanupTimer = new Timer("ActiveMQ Durable Subscriber Cleanup Timer", true);
this.cleanupTask = new TimerTask() {
@Override
public void run() {
doCleanup();
}
};
this.cleanupTimer.schedule(cleanupTask, broker.getBrokerService().getOfflineDurableSubscriberTaskSchedule(),broker.getBrokerService().getOfflineDurableSubscriberTaskSchedule());
}
}
public void doCleanup() {
long now = System.currentTimeMillis();
for (Map.Entry<SubscriptionKey, DurableTopicSubscription> entry : durableSubscriptions.entrySet()) {
DurableTopicSubscription sub = entry.getValue();
if (!sub.isActive()) {
long offline = sub.getOfflineTimestamp();
if (offline != -1 && now - offline >= broker.getBrokerService().getOfflineDurableSubscriberTimeout()) {
LOG.info("Destroying durable subscriber due to inactivity: {}", sub);
try {
RemoveSubscriptionInfo info = new RemoveSubscriptionInfo();
info.setClientId(entry.getKey().getClientId());
info.setSubscriptionName(entry.getKey().getSubscriptionName());
ConnectionContext context = new ConnectionContext();
context.setBroker(broker);
context.setClientId(entry.getKey().getClientId());
removeSubscription(context, info);
} catch (Exception e) {
LOG.error("Failed to remove inactive durable subscriber", e);
}
}
}
}
}
// The toString method for DurableTopicSubscription class
@Override
public synchronized String toString() {
return "DurableTopicSubscription-" + getSubscriptionKey() + ", id=" + info.getConsumerId() + ", active=" + isActive() + ", destinations=" + durableDestinations.size() + ", total=" + getSubscriptionStatistics().getEnqueues().getCount() + ", pending=" + getPendingQueueSize() + ", dispatched=" + getSubscriptionStatistics().getDispatched().getCount() + ", inflight=" + dispatched.size() + ", prefetchExtension=" + getPrefetchExtension();
}
评论
cleanSession
true
评论