《Java 网络编程基础》7.3 连接状态机与空闲检测

package com.plume.linkd.core;

import java.util.EnumSet;
import java.util.Map;
import java.util.Set;
import java.util.function.LongSupplier;

public final class ConnectionStateMachine {

    public enum State {
        CONNECTED, READING, CLOSING, CLOSED
    }

    private static final Map<State, Set<State>> ALLOWED = Map.of(
            State.CONNECTED, EnumSet.of(State.READING, State.CLOSING, State.CLOSED),
            State.READING, EnumSet.of(State.CONNECTED, State.CLOSING, State.CLOSED),
            State.CLOSING, EnumSet.of(State.CLOSED),
            State.CLOSED, EnumSet.noneOf(State.class));

    private final LongSupplier clock;
    private volatile State state = State.CONNECTED;
    private volatile long lastReadNanos;
    private volatile long lastWriteNanos;

    public ConnectionStateMachine() {
        this(System::nanoTime);
    }

    public ConnectionStateMachine(LongSupplier clock) {
        this.clock = clock;
        long now = clock.getAsLong();
        this.lastReadNanos = now;
        this.lastWriteNanos = now;
    }

    public State state() {
        return state;
    }

    public void onReadable() {
        State current = state;
        if (current == State.CONNECTED) {
            transition(State.READING);
        } else if (current != State.READING) {
            return;
        }
        lastReadNanos = clock.getAsLong();
    }

    public void onDrained() {
        if (state == State.READING) {
            transition(State.CONNECTED);
        }
    }

    public void touchWrite() {
        lastWriteNanos = clock.getAsLong();
    }

    public void beginClose() {
        if (state == State.CLOSED || state == State.CLOSING) {
            return;
        }
        transition(State.CLOSING);
    }

    public void closed() {
        if (state != State.CLOSED) {
            transition(State.CLOSED);
        }
    }

    public long idleNanos(long nowNanos) {
        long last = Math.max(lastReadNanos, lastWriteNanos);
        return Math.max(0, nowNanos - last);
    }

    public long idleMillis(long nowNanos) {
        return idleNanos(nowNanos) / 1_000_000L;
    }

    private void transition(State next) {
        State current = state;
        if (!ALLOWED.get(current).contains(next)) {
            throw new IllegalStateException("illegal transition " + current + " -> " + next);
        }
        state = next;
    }
}

long last = Math.max(lastReadNanos, lastWriteNanos);
long idle = Math.max(0, nowNanos - last);

package com.plume.linkd.core;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;

public final class IdleTimeoutHandler implements Handler {

    public enum Action {
        NONE, PING, CLOSE
    }

    private static final byte[] PING_FRAME = "PING\n".getBytes(StandardCharsets.US_ASCII);

    private final Handler next;
    private final ConnectionStateMachine state;
    private final long pingAfterNanos;
    private final long closeAfterNanos;
    private boolean pingSent;

    public IdleTimeoutHandler(Handler next, ConnectionStateMachine state,
                              long pingAfterMillis, long closeAfterMillis) {
        if (pingAfterMillis <= 0 || closeAfterMillis <= pingAfterMillis) {
            throw new IllegalArgumentException("expect 0 < pingAfter < closeAfter");
        }
        this.next = next;
        this.state = state;
        this.pingAfterNanos = pingAfterMillis * 1_000_000L;
        this.closeAfterNanos = closeAfterMillis * 1_000_000L;
    }

    @Override
    public void onReadable(Connection c) {
        state.onReadable();
        pingSent = false;
        next.onReadable(c);
    }

    @Override
    public void onWritable(Connection c) {
        state.touchWrite();
        next.onWritable(c);
    }

    @Override
    public void onClosed(Connection c) {
        state.closed();
        next.onClosed(c);
    }

    public Action checkIdle(Connection c, long nowNanos) {
        ConnectionStateMachine.State current = state.state();
        if (current == ConnectionStateMachine.State.CLOSED
                || current == ConnectionStateMachine.State.CLOSING) {
            return Action.NONE;
        }
        long idle = state.idleNanos(nowNanos);
        if (idle >= closeAfterNanos) {
            state.beginClose();
            c.close();
            return Action.CLOSE;
        }
        if (idle >= pingAfterNanos && !pingSent) {
            pingSent = true;
            c.enqueue(ByteBuffer.wrap(PING_FRAME.clone()));
            state.touchWrite();
            return Action.PING;
        }
        return Action.NONE;
    }

    public void halfClose(Connection c) {
        if (state.state() == ConnectionStateMachine.State.CLOSED
                || state.state() == ConnectionStateMachine.State.CLOSING) {
            return;
        }
        state.beginClose();
        try {
            c.channel().shutdownOutput();
        } catch (IOException ignored) {
            // 对端已复位或通道已关闭，交给 close() 收尾
        }
    }
}

t=0s state=CONNECTED idle=0ms
t=10s after read state=READING idle=0ms
t=20s after drain state=CONNECTED
t=81s action=PING state=CONNECTED events=[enqueue PING]
t=141s action=NONE state=CONNECTED
t=202s action=CLOSE state=CLOSING events=[enqueue PING, close]
after onClosed state=CLOSED action=NONE

Map<EventLoop, Map<Connection, IdleTimeoutHandler>> tracked = new ConcurrentHashMap<>();
ScheduledExecutorService timer = Executors.newSingleThreadScheduledExecutor(
        Thread.ofPlatform().name("linkd-idle").daemon().factory());
timer.scheduleAtFixedRate(() -> tracked.forEach((loop, perLoop) ->
        perLoop.forEach((c, handler) ->
                loop.execute(() -> handler.checkIdle(c, System.nanoTime())))),
        1, 1, TimeUnit.SECONDS);

netstat -anv -p tcp | awk '$6 == "CLOSE_WAIT" {count++} END {print count + 0}'

// 反例：墙钟被 NTP 回拨后 idle 变成负数
long idle = System.currentTimeMillis() - lastReadMillis;

// 反例：在定时线程里直接写通道
if (idle > timeout) {
    connection.channel().close();
}