Aeron集群通信异常与重启消息重放问题咨询

1. 客户端未收到PONG响应的排查调试方法 #

• 核对通道配置一致性：确认集群发送PONG的Aeron通道（如aeron:udp?endpoint=...）与客户端监听通道完全匹配，包括IP、端口、MTU参数，避免因通道不兼容导致消息无法送达。
• 校验会话活跃状态：集群侧打印会话的isActive()状态，确保发送PONG时会话处于有效状态；客户端侧检查本地会话是否完成注册，未被意外关闭。
• 开启Aeron调试日志：将io.aeron、io.aeron.cluster包日志级别设为DEBUG，重点查看：
  • 集群侧是否有sent message to session类日志，确认PONG已发出
  • 客户端侧是否存在rejected message、session not found等接收失败记录
• 网络层面抓包验证：用tcpdump或Wireshark抓取集群到客户端的UDP流量，确认PONG消息是否实际发出，以及客户端是否收到，排除防火墙、路由等网络障碍。
• 检查监听器有效性：确认客户端Subscription绑定的消息监听器实现完整，onMessage方法未因未捕获异常停止工作，且监听器已正确注册。

2. 集群重启重放所有会话消息的问题 #

2a. 快照恢复未阻止重放的原因 #

Aeron集群的快照恢复（ClusteredService#onStart）仅负责恢复集群的业务状态快照，不会截断或过滤已提交的日志条目。集群重启时重放全量未截断日志是保证状态一致性的核心机制，若你的ClusteredService未区分重放与实时场景，就会在重放时重复执行发送PONG的逻辑。
另外，若快照仅保存了业务状态，未标记哪些消息已处理无需重复响应，也会导致重放时触发重复发送。

2b. 区分重放与实时消息，仅向在线客户端发PONG #

• 利用集群内置重放标记：在ClusteredService#onSessionMessage中，通过Cluster.isReplay()判断当前消息是否为历史重放消息，若是则直接跳过发送PONG的逻辑。
• 维护会话状态快照：在onTakeSnapshot中将当前活跃会话ID列表写入快照；在onStart中从快照恢复该列表。重放消息时，仅当会话在恢复的活跃列表中且非重放场景时，才发送PONG。
• 实时跟踪会话状态：在onSessionOpened和onSessionClosed中维护内存级的活跃会话集合，重放消息时，即使目标会话曾存在，只要当前不在线就不发送响应。

核心逻辑示例代码：

private final Set<Long> activeSessions = new HashSet<>();
private Cluster cluster;

@Override
public void onStart(Cluster cluster, Image snapshotImage) {
    this.cluster = cluster;
    // 从快照恢复活跃会话列表
    if (snapshotImage != null) {
        // 读取快照中的会话ID，填充activeSessions
    }
}

@Override
public void onSessionMessage(long sessionId, long timestamp, DirectBuffer buffer, int offset, int length, Header header) {
    if (isPingMessage(buffer, offset, length)) {
        // 仅处理实时消息且会话在线的情况
        if (!cluster.isReplay() && activeSessions.contains(sessionId)) {
            cluster.sendSessionMessage(sessionId, pongBuffer, 0, pongLength);
        }
    }
}

@Override
public void onSessionOpened(long sessionId, String principal) {
    activeSessions.add(sessionId);
}

@Override
public void onSessionClosed(long sessionId, CloseReason closeReason) {
    activeSessions.remove(sessionId);
}

@Override
public void onTakeSnapshot(ExclusivePublication snapshotPublication) {
    // 将activeSessions写入快照
}

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