在Apache Mina中实现重传（NAK或超时机制）

一、客户端NAK/ACK的处理逻辑 #

处理客户端发来的NAK（请求重传）和ACK（确认接收），核心是维护已发送消息的缓存，通过消息ID关联消息内容，在对应事件中执行操作：

• 维护缓存：用IoSession属性存储已发送未确认的消息，Key为唯一消息ID，Value包含消息体、重传次数等元数据。
• 处理ACK：在messageReceived事件中识别ACK消息，根据ID移除缓存中的对应消息，同时取消该消息的超时重传任务。
• 处理NAK：同样在messageReceived事件中识别NAK消息，根据ID取出缓存消息，立即发起重传（可按需重置超时时间）。

示例代码片段：

// 自定义消息结构需包含msgId字段
@Override
public void messageReceived(IoSession session, Object message) throws Exception {
    if (message instanceof AckMessage) {
        AckMessage ack = (AckMessage) message;
        // 清理缓存与超时任务
        session.removeAttribute("SEND_CACHE_" + ack.getMsgId());
        ScheduledFuture<?> timer = (ScheduledFuture<?>) session.getAttribute("TIMER_" + ack.getMsgId());
        if (timer != null) timer.cancel(false);
    } else if (message instanceof NakMessage) {
        NakMessage nak = (NakMessage) message;
        YourMessage cachedMsg = (YourMessage) session.getAttribute("SEND_CACHE_" + nak.getMsgId());
        if (cachedMsg != null) {
            // 立即重传消息
            session.write(cachedMsg);
            // 可选：重置超时计时器
            resetTimeoutTimer(session, cachedMsg);
        }
    }
    // 业务消息处理逻辑...
}

二、超时重传机制实现 #

借助Apache Mina的IoSession.schedule()方法实现定时任务，发送消息时启动超时计时器，指定时间内未收到ACK/NAK则触发重传：

1. 发送消息时初始化缓存与超时任务：

public void sendMessage(IoSession session, byte[] payload) {
    String msgId = UUID.randomUUID().toString();
    YourMessage message = new YourMessage(msgId, payload);
    // 缓存消息
    session.setAttribute("SEND_CACHE_" + msgId, message);
    
    // 启动3秒超时任务
    ScheduledFuture<?> timer = session.schedule(() -> {
        YourMessage cachedMsg = (YourMessage) session.getAttribute("SEND_CACHE_" + msgId);
        if (cachedMsg == null) return;
        
        // 重传次数限制，避免无限循环
        if (cachedMsg.getRetryCount() >= 3) {
            session.removeAttribute("SEND_CACHE_" + msgId);
            session.closeNow(); // 超过上限关闭会话
            return;
        }
        
        // 执行重传
        session.write(cachedMsg);
        cachedMsg.incrementRetryCount();
        session.setAttribute("SEND_CACHE_" + msgId, cachedMsg);
        
        // 重置超时任务
        resetTimeoutTimer(session, cachedMsg);
    }, 3, TimeUnit.SECONDS);
    
    // 缓存超时任务，方便后续取消
    session.setAttribute("TIMER_" + msgId, timer);
}

// 重置超时任务的工具方法
private void resetTimeoutTimer(IoSession session, YourMessage message) {
    ScheduledFuture<?> oldTimer = (ScheduledFuture<?>) session.getAttribute("TIMER_" + message.getMsgId());
    if (oldTimer != null) oldTimer.cancel(false);
    
    ScheduledFuture<?> newTimer = session.schedule(() -> {
        // 复用上述超时逻辑
    }, 3, TimeUnit.SECONDS);
    session.setAttribute("TIMER_" + message.getMsgId(), newTimer);
}

2. 注意事项：
   • 必须设置重传次数上限，防止网络彻底中断时无限重传。
   • 超时时间需根据业务场景调整（弱网环境可适当延长）。

三、Apache Mina事件的有序性 #

Apache Mina对同一个IoSession的所有事件是严格有序的。IoProcessor组件会保证每个会话的事件按触发顺序串行处理，不会出现并发处理同一会话事件的情况。也就是说，messageReceived、messageSent等事件的触发顺序完全匹配消息的收发顺序，这为依赖顺序的重传机制提供了可靠基础。

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