如何解决未清理 MQTT 订阅队列
我们使用 ActiveMQ Artemis 2.8.1 和 MQTT。
我们将 ActiveMQ Artemis 与 Wildfly 集成。考虑我们的服务器使用客户端 ID SAM 连接 50 个主题。在使用 JConsole 检查 Artemis 时，我们可以看到每个客户端订阅都会生成一个名称遵循模式 <client-id>_<topic> 的队列。就我而言，将主题视为 com/api/output，这意味着订阅队列名称将为 SAM_com/api/output。同样，将有 50 个其他订阅队列使用相同的命名模式（即 SAM_<topic>）。
我的发现
根据我的研究，每个队列都用于为每个客户的订阅存储发送到该主题的消息。例如，当同一个主题（例如 1/2/3）被 3 个不同的客户端（例如 A、B、&C）订阅时，将会有 3 个订阅队列（即 A_1/2/3、B_1/2/3 和 C_1/2/3）。因此，当向主题 1/2/3 发送消息时，Artemis 会将这些消息放入订阅队列 A_1/2/3、B_1/2/3、C_1/2/3。
实际问题
现在同一个客户端想要连接到具有不同客户端 ID 的代理（例如 TOM）。我的客户端启动连接断开，Artemis 也识别连接断开，然后我的客户端连接到具有相同 50 个主题的新客户端 ID (TOM) 的代理。现在总共有 100 个订阅队列，每个主题有 2 个（即每个 clientid 一个 - SAM & TOM）。我找到了维护 SAM 队列的原因，因为在启动连接时我们使用 cleanSession 作为 false。因此，所有这些订阅队列都将是持久的，因此即使客户端断开连接，队列也会被维护。
当消息被发送到主题时，它将被放入两个队列（SAM 和 TOM）。我们的客户端通过客户端 ID TOM 连接到代理，因此 TOM 队列具有消费者，这导致客户端消耗所有 TOM 队列消息。但是，SAM 队列会累积消息并占用所有 JVM 的堆空间，直到服务器停止运行。
持久队列的目的是即使在客户端断开连接时也能维护消息，但是有没有办法告诉 ActiveMQ Artemis 如果客户端在特定时间段内没有出现或清除客户端的队列和消息当客户端断开连接时，来自客户端订阅队列的消息？
我们的 broker.xml
<configuration xmlns="urn:activemq"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:xi="http://www.w3.org/2001/XInclude"
xsi:schemaLocation="urn:activemq /schema/artemis-configuration.xsd">
<core xmlns="urn:activemq:core" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="urn:activemq:core ">
<name>0.0.0.0</name>
<security-enabled>true</security-enabled>
<persistence-enabled>false</persistence-enabled>
<!-- this could be ASYNCIO,MAPPED,NIO
ASYNCIO: Linux Libaio
MAPPED: mmap files
NIO: Plain Java Files
-->
<journal-type>ASYNCIO</journal-type>
<paging-directory>data/paging</paging-directory>
<bindings-directory>data/bindings</bindings-directory>
<journal-directory>data/journal</journal-directory>
<large-messages-directory>data/large-messages</large-messages-directory>
<journal-datasync>true</journal-datasync>
<journal-min-files>2</journal-min-files>
<journal-pool-files>10</journal-pool-files>
<journal-file-size>10M</journal-file-size>
<journal-buffer-timeout>64000</journal-buffer-timeout>
<!--
When using ASYNCIO,this will determine the writing queue depth for libaio.
-->
<journal-max-io>4096</journal-max-io>
<!--
You can verify the network health of a particular NIC by specifying the <network-check-NIC> element.
<network-check-NIC>theNicName</network-check-NIC>
-->
<!--
Use this to use an HTTP server to validate the network
<network-check-URL-list>http://www.apache.org</network-check-URL-list> -->
<!-- <network-check-period>10000</network-check-period> -->
<!-- <network-check-timeout>1000</network-check-timeout> -->
<!-- this is a comma separated list,no spaces,just DNS or IPs
it should accept IPV6
Warning: Make sure you understand your network topology as this is meant to validate if your network is valid.
Using IPs that could eventually disappear or be partially visible may defeat the purpose.
You can use a list of multiple IPs,and if any successful ping will make the server OK to continue running -->
<!-- <network-check-list>10.0.0.1</network-check-list> -->
<!-- use this to customize the ping used for ipv4 addresses -->
<!-- <network-check-ping-command>ping -c 1 -t %d %s</network-check-ping-command> -->
<!-- use this to customize the ping used for ipv6 addresses -->
<!-- <network-check-ping6-command>ping6 -c 1 %2$s</network-check-ping6-command> -->
<!-- how often we are looking for how many bytes are being used on the disk in ms -->
<disk-scan-period>5000</disk-scan-period>
<!-- once the disk hits this limit the system will block,or close the connection in certain protocols
that won't support flow control. -->
<max-disk-usage>99</max-disk-usage>
<!-- should the broker detect dead locks and other issues -->
<critical-analyzer>true</critical-analyzer>
<critical-analyzer-timeout>120000</critical-analyzer-timeout>
<critical-analyzer-check-period>60000</critical-analyzer-check-period>
<critical-analyzer-policy>HALT</critical-analyzer-policy>
<acceptors>
<!-- useEpoll means: it will use Netty epoll if you are on a system (Linux) that supports it -->
<!-- amqpCredits: The number of credits sent to AMQP producers -->
<!-- amqpLowCredits: The server will send the # credits specified at amqpCredits at this low mark -->
<!-- Note: If an acceptor needs to be compatible with HornetQ and/or Artemis 1.x clients add
"anycastPrefix=jms.queue.;multicastPrefix=jms.topic." to the acceptor url.
See https://issues.apache.org/jira/browse/ARTEMIS-1644 for more information. -->
<!-- MQTT Acceptor -->
<acceptor name="mqtt">tcp://0.0.0.0:${activeMQ-mqtt-port}?tcpSendBufferSize=1048576;tcpReceiveBufferSize=1048576;protocols=CORE,MQTT;useEpoll=true</acceptor>
<acceptor name="netty-ssl-acceptor">tcp://0.0.0.0:${activeMQ-mqtts-port}?sslEnabled=true;keyStorePath=${activeMQ-keystore-path};keyStorePassword=${activeMQ-keyStore-password};protocols=CORE,MQTT</acceptor>
</acceptors>
<security-settings>
<security-setting match="#">
<permission type="createNonDurableQueue" roles="amq"/>
<permission type="deleteNonDurableQueue" roles="amq"/>
<permission type="createDurableQueue" roles="amq"/>
<permission type="deleteDurableQueue" roles="amq"/>
<permission type="createAddress" roles="amq"/>
<permission type="deleteAddress" roles="amq"/>
<permission type="consume" roles="amq"/>
<permission type="browse" roles="amq"/>
<permission type="send" roles="amq"/>
<!-- we need this otherwise ./artemis data imp wouldn't work -->
<permission type="manage" roles="amq"/>
</security-setting>
</security-settings>
<address-settings>
<!-- if you define auto-create on certain queues,management has to be auto-create -->
<address-setting match="activemq.management#">
<dead-letter-address>DLQ</dead-letter-address>
<expiry-address>ExpiryQueue</expiry-address>
<redelivery-delay>0</redelivery-delay>
<!-- with -1 only the global-max-size is in use for limiting -->
<max-size-bytes>-1</max-size-bytes>
<message-counter-history-day-limit>10</message-counter-history-day-limit>
<address-full-policy>PAGE</address-full-policy>
<auto-create-queues>true</auto-create-queues>
<auto-create-addresses>true</auto-create-addresses>
<auto-create-jms-queues>true</auto-create-jms-queues>
<auto-create-jms-topics>true</auto-create-jms-topics>
</address-setting>
<!--default for catch all-->
<address-setting match="#">
<dead-letter-address>DLQ</dead-letter-address>
<expiry-address>ExpiryQueue</expiry-address>
<redelivery-delay>0</redelivery-delay>
<!-- with -1 only the global-max-size is in use for limiting -->
<max-size-bytes>-1</max-size-bytes>
<message-counter-history-day-limit>10</message-counter-history-day-limit>
<address-full-policy>PAGE</address-full-policy>
<auto-create-queues>true</auto-create-queues>
<auto-create-addresses>true</auto-create-addresses>
<auto-create-jms-queues>true</auto-create-jms-queues>
<auto-create-jms-topics>true</auto-create-jms-topics>
</address-setting>
</address-settings>
<addresses>
<address name="DLQ">
<anycast>
<queue name="DLQ" />
</anycast>
</address>
<address name="ExpiryQueue">
<anycast>
<queue name="ExpiryQueue" />
</anycast>
</address>
</addresses>
<broker-plugins>
<broker-plugin class-name="org.apache.activemq.artemis.core.server.plugin.impl.LoggingActiveMQServerPlugin">
<property key="LOG_ALL_EVENTS" value="true"/>
<property key="LOG_CONNECTION_EVENTS" value="true"/>
<property key="LOG_SESSION_EVENTS" value="true"/>
<property key="LOG_CONSUMER_EVENTS" value="true"/>
<property key="LOG_DELIVERING_EVENTS" value="true"/>
<property key="LOG_SENDING_EVENTS" value="true"/>
<property key="LOG_INTERNAL_EVENTS" value="true"/>
</broker-plugin>
</broker-plugins>
</core>
</configuration>
解决方法
订阅完成后是否真的要删除订阅。换句话说，MQTT 客户端应该在创建新订阅者之前取消订阅现有订阅者。代理无法知道订阅者是否计划稍后重新连接以从其订阅中获取消息。
如果您真的希望经纪人删除订阅队列，您可以使用以下地址设置：
-
<auto-delete-queues>：这必须是true。 -
<auto-delete-queues-delay>：这是从最后一个消费者断开连接到代理删除订阅队列的延迟（以毫秒为单位）。 -
<auto-delete-queues-message-count>：必须是-1才能忽略订阅队列中的任何消息。
明确地说，我建议反对这种配置，因为代理可能会无意中删除合法消息。例如，如果订阅者错误地断开连接（例如，由于网络故障、硬件故障、JVM 崩溃等）的时间超过配置的 <auto-delete-queues-delay>，则代理将删除其订阅队列，并且所有订阅者的消息将有效迷路了。
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 dio@foxmail.com 举报，一经查实，本站将立刻删除。
