java

《Netty 高性能网络编程》8.1 客户端连接与重连

客户端重连的难点不在 connect 调用，而在失败路径、触发时机与退避节奏。本节拆开 Bootstrap.connect 的拒绝、超时与握手失败三种结果，用 closeFuture 触发重连，给出 linkd-v2 ReconnectHandler 的状态机、指数退避、0.8 到 1.2 抖动、排队上限与惊群规避。

Leeting Yan 2026-10-28 8 分钟阅读 3712 字

本节目标：让 linkd-v2 客户端在服务端重启、握手失败、网络中断时按可控节奏恢复，而不是把重试塞进一个 while 循环。
适用版本：JDK 25 + Netty 4.2.19.Final。

承接的问题

7.3 的客户端在 channelActive 时发出 HELLO，收到 HELLO_ACK 后才允许发送业务帧；RpcClient 用 requestId 把响应关联回调用方。服务端不重启时这条链路工作正常，一旦 9093 背后的进程退出，客户端只会看到一次 channelInactive，之后所有 call() 都停在等待响应的 future 上，直到调用方超时。

卷一第 3 章的 v1 客户端用 while (running) 包住 new Socket(...)，catch (IOException) 里 Thread.sleep(3000) 再重来。它有两个问题。第一，sleep 占住一个平台线程，连接目标一多，线程数就跟着涨。第二，固定 3 秒让所有客户端在同一时刻重试，服务端刚恢复就被一整批 SYN 和握手打满。

Netty 的客户端不能照搬这个循环：Bootstrap.connect() 本身是异步的，节奏应该交给 EventLoop 的定时任务，调用线程只负责提交意图。这一节只解决连接本身：connect() 失败有哪几种结果、什么时候该触发重连、退避怎么算、断线期间请求往哪里放。8.2 把单连接换成一个有上限的连接池，8.3 处理关闭、半关闭与回收。

本机口径：Apple M1 Pro，10 核，32 GB，macOS 26.3 (25D125)；Temurin 25.0.5+8，Netty 4.2.19.Final。下面没有性能数字，异常文本以实际运行为准。

核心机制一：connect() 的失败路径与监听器

Bootstrap.connect(host, port) 返回 ChannelFuture，它在三种情况下完成：TCP 建连成功、建连被拒绝、建连超时。DNS 解析失败也会让 future 失败，但原因类型不同。失败时 Netty 会关闭这条 channel，future.channel() 仍然非 null，只是不再 active；失败原因在 future.cause()。

Bootstrap bootstrap = new Bootstrap()
        .group(eventLoop)
        .channel(NioSocketChannel.class)
        .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 2000)
        .option(ChannelOption.TCP_NODELAY, true)
        .handler(initializer);

ChannelFuture connect = bootstrap.connect("127.0.0.1", 9093);
connect.addListener((ChannelFutureListener) future -> {
    if (future.isSuccess()) {
        LOG.log(System.Logger.Level.INFO, "tcp connected, local={0}",
                future.channel().localAddress());
    } else {
        LOG.log(System.Logger.Level.WARNING, "connect failed: {0}",
                future.cause().toString());
    }
});

几种典型失败，示意输出（异常文本随平台和网络栈变化）：

WARN  connect failed: java.net.ConnectException: Connection refused
WARN  connect failed: io.netty.channel.ConnectTimeoutException: connection timed out: /127.0.0.1:9093
WARN  connect failed: java.net.UnknownHostException: linkd.internal

失败原因	触发条件	是否值得立即重试
`ConnectException`	端口没有监听，收到 RST	是，但要退避
`ConnectTimeoutException`	SYN 无响应，超过 `CONNECT_TIMEOUT_MILLIS`	是，退避更长
`UnknownHostException`	域名解析失败	是，但应记录并告警

CONNECT_TIMEOUT_MILLIS 要显式设置。不设置时默认值偏长，服务端已经挂掉的情况下，一次 connect() 会占用一个 EventLoop 任务很久；在 2 秒内失败并进入退避，比等十几秒更有意义。

监听器在 channel 所属的 EventLoop 上执行，不是调用 connect() 的线程。因此监听器里可以直接读写 channel 状态，但不允许 sleep、JDBC、文件 IO，也不允许 sync()。

核心机制二：closeFuture 是重连的触发点

建连成功只是第一步，连接可能在任意时刻被对端关闭。channel.closeFuture() 在这个 channel 关闭时完成，是挂重连逻辑的正确位置：

if (!future.isSuccess()) {
    scheduleReconnect("connect failed");
    return;
}
future.channel().closeFuture().addListener(closed -> onChannelClosed(future.channel()));

这里有一个容易踩的坑：连接失败时 channel 已经被 Netty 关闭，closeFuture() 也已经完成。如果既在失败分支调 scheduleReconnect，又无条件挂 closeFuture 监听器，就会排两次重连，最终建出两条连接。规则是二选一：失败分支直接排下一次，成功分支才挂 closeFuture。

closeFuture 也不区分是谁关的。服务端发 FIN、本地调用 close()、8.3 的优雅关闭，都会让它完成。重连逻辑必须能识别「主动关闭」：shutdown() 先把状态置为 CLOSING，closeFuture 回调看到 CLOSING 就直接返回，不再排下一次。

核心机制三：指数退避、抖动与上限

退避的目标不是「等得越久越安全」，而是让重试次数在时间上摊开，同时给恢复留出快速通道。linkd-v2 用 200 ms 起步、每次翻倍、10 秒封顶，再乘一个 0.8–1.2 的随机因子：

private static long nextDelayMillis(int attempt) {
    int shift = Math.min(attempt - 1, 6);
    long base = Math.min(200L << shift, 10_000L);
    return (long) (base * ThreadLocalRandom.current().nextDouble(0.8, 1.2));
}

scheduled = eventLoop.schedule(this::connectOnce, delay, TimeUnit.MILLISECONDS);

第几次尝试	基准延迟	抖动后区间
1	200 ms	160–240 ms
2	400 ms	320–480 ms
3	800 ms	640–960 ms
4	1600 ms	1280–1920 ms
5	3200 ms	2560–3840 ms
6 及以后	6400–10000 ms	5120–12000 ms

三个细节：

用 EventLoop.schedule，不用 Thread.sleep。 定时任务挂在 channel 所属的事件循环上，不占额外线程；shutdown() 可以 cancel 它，sleep 中的线程没有这个能力。
成功归零。 只有收到 HELLO_ACK、状态进入 READY 时才把 attempt 清零。TCP 建连成功就清零，会让握手持续失败的目标一直以 200 ms 的间隔重试。
抖动在客户端各自计算。 一千个客户端即使同时启动，ThreadLocalRandom 也会把它们摊到 0.8–1.2 倍区间里，服务端不会在同一毫秒收到一千条连接。这是避免惊群最省事的手段，比在服务端加 SO_BACKLOG 更直接。

在 linkd-v2 里落地：ReconnectHandler 与状态机

ReconnectHandler 把连接状态限制在四种：DISCONNECTED、CONNECTING、READY、CLOSING。所有状态迁移都在同一个 EventLoop 上执行，因此字段不需要加锁；只有排队计数用原子变量，因为 send() 可能从业务线程调用。

当前状态	事件	下一状态	动作
`DISCONNECTED`	`reconnect()`	`CONNECTING`	发起 `connect()`
`CONNECTING`	TCP 成功	`CONNECTING`	挂 `closeFuture`，等握手
`CONNECTING`	TCP 失败	`DISCONNECTED`	`attempt++`，排下一次
`CONNECTING`	`markReady()`	`READY`	`attempt=0`，排空队列
`CONNECTING`	channel 关闭	`DISCONNECTED`	`attempt++`，排下一次
`READY`	channel 关闭	`DISCONNECTED`	归零后按第 1 次退避重连
任意	`shutdown()`	`CLOSING`	取消定时器、关连接、清队列

package com.plume.linkd.v2.client;

import com.plume.linkd.v2.protocol.LinkdFrame;
import com.plume.linkd.v2.protocol.ProtocolConstants;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import java.util.ArrayDeque;
import java.util.Deque;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public final class ReconnectHandler {

    public enum State { DISCONNECTED, CONNECTING, READY, CLOSING }

    private record Pending(LinkdFrame frame, Runnable onAbandoned, long deadlineNanos) {}

    private static final System.Logger LOG = System.getLogger(ReconnectHandler.class.getName());
    private static final int MAX_PENDING_FRAMES = 1024;
    private static final int MAX_PENDING_BYTES = 8 << 20;

    private final String host;
    private final int port;
    private final EventLoop eventLoop;
    private final Bootstrap bootstrap;
    private final Deque<Pending> pending = new ArrayDeque<>();
    private final AtomicInteger pendingFrames = new AtomicInteger();
    private final AtomicInteger pendingBytes = new AtomicInteger();

    private State state = State.DISCONNECTED;
    private int attempt;
    private Channel channel;
    private ScheduledFuture<?> scheduled;

    public ReconnectHandler(String host, int port, EventLoop eventLoop,
                            ChannelInitializer<SocketChannel> initializer) {
        this.host = host;
        this.port = port;
        this.eventLoop = eventLoop;
        this.bootstrap = new Bootstrap()
                .group(eventLoop)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 2000)
                .option(ChannelOption.TCP_NODELAY, true)
                .handler(initializer);
    }

    public void reconnect() {
        eventLoop.execute(() -> {
            if (state == State.DISCONNECTED) {
                connectOnce();
            }
        });
    }

    public boolean send(LinkdFrame frame, Runnable onAbandoned, long timeoutMillis) {
        int bytes = ProtocolConstants.HEADER_LENGTH + frame.payload().length;
        int frames = pendingFrames.incrementAndGet();
        int bytesNow = pendingBytes.addAndGet(bytes);
        if (frames > MAX_PENDING_FRAMES || bytesNow > MAX_PENDING_BYTES) {
            pendingFrames.decrementAndGet();
            pendingBytes.addAndGet(-bytes);
            return false;
        }
        long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(timeoutMillis);
        try {
            eventLoop.execute(() -> {
                Pending p = new Pending(frame, onAbandoned, deadline);
                if (state == State.READY) {
                    write(p, bytes);
                } else {
                    pending.addLast(p);
                }
            });
        } catch (RejectedExecutionException e) {
            pendingFrames.decrementAndGet();
            pendingBytes.addAndGet(-bytes);
            return false;
        }
        return true;
    }

    public void markReady(Channel readyChannel) {
        eventLoop.execute(() -> {
            if (readyChannel != channel || state != State.CONNECTING) {
                return;
            }
            state = State.READY;
            attempt = 0;
            Pending p;
            while ((p = pending.pollFirst()) != null) {
                write(p, ProtocolConstants.HEADER_LENGTH + p.frame().payload().length);
            }
        });
    }

    public void shutdown() {
        eventLoop.execute(() -> {
            if (state == State.CLOSING) {
                return;
            }
            state = State.CLOSING;
            if (scheduled != null) {
                scheduled.cancel(false);
            }
            if (channel != null) {
                channel.close();
            }
            Pending p;
            while ((p = pending.pollFirst()) != null) {
                pendingFrames.decrementAndGet();
                pendingBytes.addAndGet(-ProtocolConstants.HEADER_LENGTH - p.frame().payload().length);
                p.onAbandoned().run();
            }
        });
    }

    private void connectOnce() {
        if (state != State.DISCONNECTED) {
            return;
        }
        state = State.CONNECTING;
        bootstrap.connect(host, port).addListener((ChannelFutureListener) future -> {
            if (state != State.CONNECTING) {
                future.channel().close();
            } else if (future.isSuccess()) {
                onTcpConnected(future.channel());
            } else {
                LOG.log(System.Logger.Level.WARNING, "connect failed: {0}",
                        future.cause().toString());
                state = State.DISCONNECTED;
                scheduleReconnect();
            }
        });
    }

    private void onTcpConnected(Channel ch) {
        channel = ch;
        ch.closeFuture().addListener(closed -> {
            if (ch != channel) {
                return;
            }
            channel = null;
            if (state == State.CLOSING) {
                return;
            }
            if (state == State.READY) {
                attempt = 0;
            }
            state = State.DISCONNECTED;
            scheduleReconnect();
        });
    }

    private void scheduleReconnect() {
        if (state == State.CLOSING) {
            return;
        }
        attempt++;
        long delay = nextDelayMillis(attempt);
        LOG.log(System.Logger.Level.INFO, "reconnect in {0} ms (attempt {1})", delay, attempt);
        scheduled = eventLoop.schedule(this::connectOnce, delay, TimeUnit.MILLISECONDS);
    }

    private void write(Pending p, int bytes) {
        pendingFrames.decrementAndGet();
        pendingBytes.addAndGet(-bytes);
        if (System.nanoTime() - p.deadlineNanos() >= 0) {
            p.onAbandoned().run();
            return;
        }
        channel.writeAndFlush(p.frame()).addListener((ChannelFutureListener) f -> {
            if (!f.isSuccess()) {
                f.channel().close();
            }
        });
    }

    private static long nextDelayMillis(int attempt) {
        int shift = Math.min(attempt - 1, 6);
        long base = Math.min(200L << shift, 10_000L);
        return (long) (base * ThreadLocalRandom.current().nextDouble(0.8, 1.2));
    }
}

几个关键点：

状态只在 EventLoop 上改。 reconnect()、markReady()、shutdown() 都先 eventLoop.execute，把调用折叠到同一个线程；重复调用 reconnect() 时，后续任务看到 CONNECTING 或 READY 直接返回，天然幂等。
markReady 校验 channel 身份。 握手期间可能已经发生一次断开并重连，旧 channel 的 HELLO_ACK 迟到时不能把新连接置为可用；readyChannel != channel 直接丢弃。
排队有双上限。 1024 帧和 8 MiB 两个阈值任一超出就返回 false，调用方用 RejectedExecutionException 结束 future。只限帧数不够：单帧负载上限 1 MiB，1024 帧最坏能占 1 GiB 堆。
onAbandoned 覆盖两种结局。 排队期间超时、或 shutdown() 清队列，都会回调它；RpcClient 用它把等待中的 future 以 TimeoutException 结束，避免调用方一直等。
send 在 READY 时也走 EventLoop。 业务线程直接 writeAndFlush 会与队列里的帧竞争顺序；统一入队后由 write 写，顺序与调用顺序一致。

RpcClient 把客户端握手 handler 构造成 new ClientHandshakeHandler(reconnect::markReady) 装进 pipeline：channelActive 时发 HELLO，收到 HELLO_ACK 后把 READY 属性置 true 并回调 markReady。把就绪信号做成回调，ReconnectHandler 就不依赖具体握手 handler 类型。

常见错误与边界

错误一：在监听器里 Thread.sleep 或 sync()。

// 反例：阻塞 EventLoop
connect.addListener(f -> {
    if (!f.isSuccess()) {
        Thread.sleep(1000);
        bootstrap.connect(host, port);
    }
});

监听器运行在 EventLoop 线程上，sleep 会暂停这个线程上所有 channel 的读写；这段时间里其他连接的心跳、响应全部停摆。用 eventLoop.schedule。

错误二：失败路径和 closeFuture 同时排重连。

// 反例：失败时 channel 已被关闭，closeFuture 立即回调
connect.addListener(f -> scheduleReconnect());
connect.channel().closeFuture().addListener(f -> scheduleReconnect());

一次失败会排两次重连，且两次都可能成功，结果是同一目标出现两条连接、两套 requestId 空间。要么只在失败分支排，要么只在 closeFuture 排，不要两处都挂。

错误三：TCP 建连成功就重置退避。

握手认证失败时，TCP 是通的，但对端每个连接都会拒绝。若在 connect 成功时清零 attempt，客户端会以 200 ms 的节奏无限重连，日志被刷满，服务端认证模块持续承压。重置点必须是 HELLO_ACK 之后。

小结

Bootstrap.connect() 的失败原因要分类记录：拒绝、超时、解析失败的处理节奏不同，不能一律立即重试。
连接失败时 channel 已被 Netty 关闭；重连触发点只能在失败监听器和成功后的 closeFuture 中二选一。
退避用 EventLoop.schedule，200 ms 起步、翻倍、10 秒封顶并乘 0.8–1.2 抖动；计数只在 HELLO_ACK 后清零。
ReconnectHandler 的四个状态都在同一个 EventLoop 上迁移，重复调用 reconnect() 幂等；断线期间排队同时限制帧数和字节数，超限快速失败。

单连接能稳定重连之后，下一个问题是并发：一条连接上同时能压多少请求、什么时候需要多条连接、连接池按什么粒度借还。8.2 用 FixedChannelPool 和 AbstractChannelPoolMap 回答这些问题。

阅读导航：上一节：7.3 握手、认证与连接绑定 · 下一节：8.2 连接池与多路复用。

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章返回首页