ActiveMQ Classic STOMP持久化订阅断开连接时消息丢失问题

核心问题根源 #

你遇到的问题本质是TCP连接关闭时机不当，以及STOMP协议交互逻辑的疏漏，导致ActiveMQ已发送的MESSAGE/RECEIPT帧在客户端TCP连接提前关闭时丢失；同时持久化订阅的状态同步问题，使得未接收的消息无法在重连后被重新投递。

客户端逻辑优化（优先解决） #

1. 替换“超时判断无消息”为“主动取消订阅” #

不要依赖超时来判断没有剩余消息，改为：

• 在步骤3接收完当前可用消息后，发送UNSUBSCRIBE帧（指定订阅ID）取消持久化订阅
• 等待UNSUBSCRIBE的回执（如果请求了），确认ActiveMQ已停止向该连接发送消息
• 再执行步骤4的DISCONNECT操作

这样能确保ActiveMQ不会在DISCONNECT流程中再发送新的MESSAGE帧，从源头避免消息和回执的冲突。

2. 严格保证DISCONNECT的回执等待逻辑 #

• 发送DISCONNECT帧时必须指定receipt头部，明确要求回执
• 步骤5的接收逻辑不要设置固定超时，改为无限等待直到收到对应RECEIPT帧，或设置一个足够长的超时（比如30秒，根据你的消息传输延迟调整）
• 只有收到RECEIPT帧后，再主动关闭TCP连接，避免TCP提前断开导致已发送的帧丢失

3. 规范MESSAGE帧的ACK处理 #

根据你订阅时指定的ack模式：

• 如果是client模式：必须对每一个收到的MESSAGE帧发送ACK帧，确保ActiveMQ知道消息已被接收；未ACK的消息会在重连后重新投递
• 如果是auto模式：确认ActiveMQ的stomp.autoack配置为默认的true，避免因自动ACK失效导致消息被误标记为已处理

4. 固定客户端ID与订阅ID #

持久化订阅必须保证客户端ID（client-id头部）和订阅ID（id头部）全局唯一且固定，否则重连时会创建新的订阅，旧的未投递消息会绑定到原订阅，新订阅无法接收。

ActiveMQ配置调整（辅助优化） #

1. 禁用Nagle算法 #

在activemq.xml的transportConnectors中添加stomp.disable.nagle=true，减少TCP小帧的延迟发送，避免RECEIPT帧被延迟：

<transportConnector name="stomp" uri="stomp://0.0.0.0:61613?stomp.disable.nagle=true"/>

2. 调整TCP缓冲区大小 #

根据消息体积调整发送/接收缓冲区，避免数据溢出：

<transportConnector name="stomp" uri="stomp://0.0.0.0:61613?stomp.sendBufferSize=65536&stomp.receiveBufferSize=65536"/>

3. 优化连接空闲超时 #

调整maxInactivityDuration和maxInactivityDurationInitalDelay，避免ActiveMQ在客户端处理流程中主动断开连接：

<transportConnector name="stomp" uri="stomp://0.0.0.0:61613?maxInactivityDuration=300000&maxInactivityDurationInitalDelay=60000"/>

4. 启用持久化消息的重投递检查 #

确保activemq.xml中persistenceAdapter的useJournal和useQuickJournal为true，保证未投递的持久化消息不会丢失：

<persistenceAdapter>
    <kahaDB directory="${activemq.data}/kahadb" useJournal="true" useQuickJournal="true"/>
</persistenceAdapter>

验证方案 #

调整后测试流程：

1. 创建TCP连接，指定固定client-id
2. 发送SUBSCRIBE帧，指定固定id和正确的ack模式
3. 循环接收MESSAGE帧并处理/ACK，直到没有新消息（可短暂等待1-2秒，而非依赖超时）
4. 发送UNSUBSCRIBE帧，等待回执
5. 发送DISCONNECT帧（带receipt），等待RECEIPT帧
6. 收到RECEIPT后关闭TCP连接

重连时使用相同的client-id和id，验证未接收的消息是否能被重新投递。

内容的提问来源于stack exchange，提问作者Tim