在Amazon MQ（ActiveMQ）故障转移中避免消息丢失并保障高吞吐量

核心思路 #

要平衡高吞吐量和故障转移时的消息可靠性，关键是在异步发送的前提下，确保消息在被确认前已完成持久化到EFS，同时避免阻塞发送线程。

具体解决方案 #

1. 异步发送+持久化确认回调 #

不要单纯依赖setUseAsyncSend(true)的无确认异步发送，改用带回调的异步发送API，在消息成功持久化到EFS后再收到确认信号。这种方式既保留了异步发送的高吞吐量，又能明确知道哪些消息已持久化，未收到确认的消息可在故障后重发。

示例代码（Java）：

// 初始化生产者时开启异步发送
producer.setUseAsyncSend(true);

// 发送消息时指定异步回调
producer.sendAsync(message, new AsyncCallback() {
    @Override
    public void onSuccess() {
        // 消息已成功持久化到EFS，可记录日志或清理本地缓存
    }

    @Override
    public void onException(JMSException exception) {
        // 消息持久化失败，可将消息加入本地重试队列，后续异步重发
    }
});

• 优势：发送线程无需等待同步确认，吞吐量不受影响；仅对未确认的消息做重发，避免无效开销。

2. 优化Amazon MQ代理的持久化配置 #

通过自定义Amazon MQ代理的activemq.xml配置，调整KahaDB（默认持久化引擎）的异步写入策略，减少内存中未持久化消息的积压：

• 设置kahaDB.asyncWriteEnabled=true：开启异步写入EFS，提升持久化效率
• 设置kahaDB.syncOnWrite=false：避免每次写入都强制同步到磁盘（需配合回调确认使用，否则仍有丢失风险）
• 调整persistentDeliveryPolicy的asyncSendMaxPendingMessagesPerConnection：限制每个连接的异步待确认消息数，防止内存积压过多无法及时持久化

示例配置片段：

<persistenceAdapter>
    <kahaDB directory="${activemq.data}/kahadb" 
            asyncWriteEnabled="true" 
            syncOnWrite="false"/>
</persistenceAdapter>
<policyEntry queue=">" sendAsyncAcks="true">
    <persistentDeliveryPolicy asyncSendMaxPendingMessagesPerConnection="1000"/>
</policyEntry>

• 优势：从代理层面优化持久化效率，减少内存中待持久化消息的数量，降低故障转移时的丢失概率，且不影响发送吞吐量。

3. 客户端侧实现幂等性+异步重试机制 #

即使出现少量未持久化的消息丢失，通过客户端的幂等性保证和异步重试，可实现最终一致性：

• 为每条消息生成唯一业务标识（如订单ID、消息UUID），消费端通过该标识去重

• 客户端维护一个本地内存队列，存放未收到确认的消息，后台启动独立线程批量重发这些消息，不阻塞主线程的发送流程

• 优势：兼容极端情况下的消息丢失，且重试操作异步执行，不会影响主发送流程的吞吐量。

4. 启用Amazon MQ主备的同步持久化模式 #

如果业务对可靠性要求极高，可在Amazon MQ代理配置中启用主备同步持久化：即主节点在返回发送确认前，确保消息已同步到备节点的EFS存储中。此模式会略微增加延迟，但仍远低于同步发送的阻塞模式，吞吐量下降幅度可控（尤其在EFS性能充足的情况下）。

性能测试注意事项 #

• 模拟真实故障场景（如主动重启主节点），统计未确认消息的数量和重发成功率
• 监控EFS的IOPS和延迟，确保持久化效率能跟上发送速度，避免内存积压
• 调整asyncSendMaxPendingMessagesPerConnection参数，找到吞吐量和内存占用的平衡点

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