当 WebSocket 遇上多实例部署：Netty 原生长连接与网关粘性路由实战复盘-阿里云开发者社区

本文是「成都硅基边界」零代码构建平台技术复盘系列，继续 Java 侧业务后端（Silicon-Service）。上一篇聊了 JWT + Redis 双层鉴权，这一篇聊长连接：为什么 WebSocket 一旦上了多实例部署就全是坑，以及我们怎么用"网关粘性路由 + Netty 原生服务"把这两个坑填平。

文中代码均为示意代码（根据设计改写、非项目真实源码），按生产级 Java 规范书写。

一、背景：长连接在分布式下会遇到什么

HTTP 请求是"短平快"的：来一个请求、做完就走，随便打到哪个实例都行——所以负载均衡可以肆无忌惮地轮询。

WebSocket 完全不是这样：连接一旦建立就绑死在某个实例上，后续所有消息都必须由这个实例收发。于是多实例部署下立刻冒出两个难题：

难题一：连接该往哪个实例路由？ 默认轮询负载均衡会把同一个客户端的重连请求分到不同实例，导致"连接在实例 A、消息发到实例 B"——推送永远到不了。
难题二：流式消息为什么会乱序？ LLM 应用的回答是流式分片推送的（打字机效果）。多个线程同时往同一个连接写数据，到了 Netty 的 EventLoop 里就可能重排——"结束标记"比最后一段正文先到，前端直接错乱。

我们最终落地的方案是：网关注入粘性路由（解决难题一）+ Netty 原生独立服务（解决难题二）。整体链路如下：

flowchart LR
    C[客户端<br/>网页客服 / 管理端对话] -->|ws://.../ws/chat<br/>?token=xxx| GW[网关 :8080]
    GW --> SF[WsStickyRoutingFilter<br/>order=10050]
    SF -->|按客户端 IP 一致性哈希<br/>替换 lb:// 为具体实例| N1[agent-service A<br/>Netty :27051]
    SF -.->|实例下线则重分配| N2[agent-service B<br/>Netty :27051]
    N1 --> H[NioWebSocketHandler<br/>连接表 / 账号绑定]
    H --> SE[单线程 STREAM_EXECUTOR<br/>保证流式分片保序]
    SE -->|TextWebSocketFrame| C

二、难题一：网关粘性路由

2.1 为什么不能用默认负载均衡

Spring Cloud Gateway 默认会把 lb://agent-service/** 交给 ReactiveLoadBalancerClientFilter 做轮询。对 WebSocket 来说这等于"每次都随机换一个实例"，连接和后续消息必然对不上。

我们要做的是：在负载均衡之前，把 lb:// 直接替换成具体实例地址，让它跳过默认 LB。

2.2 关键在于过滤器顺序

网关的过滤器靠 Ordered 决定执行顺序，这里存在一个精确的"夹缝"：

RouteToRequestUrlFilter           order = 10000   ← 把路由解析成 lb://agent-service/ws/chat
        ↓  ★ 我们的粘性路由插在这里（10050）★
ReactiveLoadBalancerClientFilter  order = 10150   ← 默认负载均衡（要被跳过）

只有卡在这两者之间，才能拿到已经解析成 lb:// 的 URL，并且在默认 LB 生效前把它改掉。实现如下：

/**
 * WebSocket 粘性路由过滤器：按客户端标识把 WS 请求固定转发到同一实例。
 * <p>执行时机在 RouteToRequestUrlFilter(10000) 之后、ReactiveLoadBalancerClientFilter(10150)
 * 之前，通过替换 GATEWAY_REQUEST_URL_ATTR 跳过默认负载均衡。
 */
@Component
@RequiredArgsConstructor
@Slf4j
public class WsStickyRoutingFilter implements GlobalFilter, Ordered {
   

    private static final String TARGET_SERVICE = "agent-service";
    private static final String WS_PATH_PREFIX = "/ws";
    /** agent-service 独立 Netty WebSocket 端口（与 HTTP 端口不同，需替换注册端口） */
    private static final int WS_PORT = 27051;

    private final ReactiveDiscoveryClient discoveryClient;

    /** 客户端标识 -> 实例地址(host:port) 的粘性映射缓存 */
    private final Map<String, String> stickyMap = new ConcurrentHashMap<>();

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
   
        URI url = exchange.getAttribute(GATEWAY_REQUEST_URL_ATTR);
        // 只处理 lb://agent-service/ws/** 这一类请求
        if (url == null || !"lb".equals(url.getScheme())
                || !TARGET_SERVICE.equals(url.getHost())
                || !url.getPath().startsWith(WS_PATH_PREFIX)) {
            return chain.filter(exchange);
        }
        String clientKey = resolveClientKey(exchange);
        return discoveryClient.getInstances(TARGET_SERVICE)
                .collectList()
                .flatMap(instances -> {
                    if (instances.isEmpty()) {
                        log.error("粘性路由：无可用实例，回落默认负载均衡");
                        return chain.filter(exchange);
                    }
                    String address = resolveInstance(clientKey, instances);
                    URI directUri = URI.create(
                            url.toString().replace("lb://" + TARGET_SERVICE, "http://" + address));
                    exchange.getAttributes().put(GATEWAY_REQUEST_URL_ATTR, directUri);
                    return chain.filter(exchange);
                });
    }

    /**
     * 解析目标实例：优先复用缓存，缓存实例已下线则重新分配。
     */
    private String resolveInstance(String clientKey, List<ServiceInstance> instances) {
   
        String cached = stickyMap.get(clientKey);
        if (cached != null) {
   
            boolean alive = instances.stream()
                    .anyMatch(i -> (i.getHost() + ":" + WS_PORT).equals(cached));
            if (alive) {
   
                return cached;
            }
            stickyMap.remove(clientKey);           // 实例下线，重新分配
            log.info("粘性路由：实例已下线，重新分配 client={} old={}", clientKey, cached);
        }
        ServiceInstance selected = instances.get(Math.abs(clientKey.hashCode()) % instances.size());
        String address = selected.getHost() + ":" + WS_PORT;
        stickyMap.put(clientKey, address);
        log.info("粘性路由：新分配 client={} -> {}", clientKey, address);
        return address;
    }

    /** 取客户端标识：优先代理头，逐级回落到真实远端地址 */
    private String resolveClientKey(ServerWebExchange exchange) {
   
        HttpHeaders headers = exchange.getRequest().getHeaders();
        String forwarded = headers.getFirst("X-Forwarded-For");
        if (StringUtils.isNotBlank(forwarded)) {
   
            return forwarded.split(",")[0].trim();
        }
        String realIp = headers.getFirst("X-Real-IP");
        if (StringUtils.isNotBlank(realIp)) {
   
            return realIp;
        }
        InetSocketAddress remote = exchange.getRequest().getRemoteAddress();
        return remote != null ? remote.getAddress().getHostAddress() : "unknown";
    }

    @Override
    public int getOrder() {
   
        return 10050;      // 必须介于 10000 与 10150 之间
    }
}

2.3 两个必须知道的局限

hashCode % 实例数 在扩缩容时会大面积重映射：实例数从 3 变 4，几乎所有客户端的映射都会变，老实例上的连接被迫全部重连。规模上来后应换成一致性哈希环 + 虚拟节点，把重映射控制在 1/N；
同 NAT 出口的多用户会挤到同一实例：公司内网几百个用户可能共用一个出口 IP，按 IP 哈希会把它们全压到一台机器上。更稳的做法是用"用户/会话 ID"而不是 IP 做键（鉴权后即可拿到）。

三、难题二：流式消息为什么必须保序

3.1 根因在 EventLoop 的 MPSC 队列

Netty 的 Channel.writeAndFlush() 是线程安全的——你可以从任意线程调用。但"线程安全"不等于"全序保证"：多个线程并发提交写任务时，任务进入 EventLoop 的 MPSC（多生产者单消费者）队列，入队顺序取决于线程调度的先后，并不能保证与你业务代码的调用顺序一致。

对流式回答来说这是致命的：

业务线程顺序：  分片1 → 分片2 → 分片3 → 结束(end=true)
EventLoop 实际：分片1 → 分片3 → 结束 → 分片2
前端看到：      "你好，" + "世界" + [结束] + "，欢迎"

"结束标记"提前到达，前端会把回答提前截断，用户看到半句话就停了。

3.2 解法：单线程串行化写入

我们给流式消息单独开了一个单线程执行器，所有流式分片都提交到这一个线程里按序发送：

/**
 * 流式消息专用单线程执行器。
 * <p>保证 streamMsg 严格按调用顺序串行写入 Netty Channel，
 * 避免多线程 writeAndFlush 提交到 EventLoop MPSC 队列时发生乱序
 * （表现为 end=true 的分片先于正文分片到达前端，前端提前终止渲染）。
 */
private static final ExecutorService STREAM_EXECUTOR = Executors.newSingleThreadExecutor(r -> {
   
    Thread t = new Thread(r, "ws-stream-sender");
    t.setDaemon(true);          // 守护线程，不阻塞应用退出
    return t;
});

/**
 * 推送流式分片（按序）。
 *
 * @param customerId 目标客户（可能对应多个连接：多标签页）
 * @param msgId      流式消息 ID（前端据此归并同一段回答）
 * @param text       本片文本
 * @param end        是否为最后一片
 */
public static boolean pushStreamChunk(Long customerId, String msgId, String text, boolean end) {
   
    try {
   
        STREAM_EXECUTOR.execute(() -> {
   
            StreamMessage payload = new StreamMessage(msgId, text, end);
            broadcastToCustomer(customerId, ServerMessage.of(MessageType.streamMsg, payload));
        });
        return true;
    } catch (Exception e) {
   
        log.error("流式消息推送失败 customerId={} msgId={}", customerId, msgId, e);
        return false;
    }
}

这个设计的本质是"用一次线程切换换取顺序保证"。有人会担心单线程成为瓶颈——对文本流式分片来说完全不必：分片体积小（几十字节到几 KB），发送是内存写 + 网络缓冲，单线程每秒可处理数万条；而顺序正确性对此场景是硬需求。如果未来单线程真的成为瓶颈，正确方向是"每连接一个有序队列"而不是"多线程共享连接"。

四、Netty 服务搭建的四个工程细节

4.1 异步启动，不拖慢应用启动

Netty 服务放在独立的初始化组件里，通过线程池异步启动，避免 bind() 阻塞 Spring 启动流程：

@Component
@RequiredArgsConstructor
@Slf4j
public class WebSocketServerBootstrap implements InitializingBean {
   

    private static final int WS_PORT = 27051;

    private final WebSocketChannelInitializer channelInitializer;
    @Resource(name = "taskExecutor")
    private ThreadPoolTaskExecutor taskExecutor;

    @Override
    public void afterPropertiesSet() {
   
        taskExecutor.execute(this::start);      // 异步启动，不阻塞 Spring 容器
    }

    private void start() {
   
        // bossGroup 只负责接受连接，1 个线程足够；workerGroup 负责 IO 读写，默认 CPU * 2
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
   
            ServerBootstrap bootstrap = new ServerBootstrap()
                    .group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .handler(new LoggingHandler(LogLevel.INFO))     // 入/出站事件日志，便于排查
                    .childHandler(channelInitializer);
            Channel channel = bootstrap.bind(WS_PORT).sync().channel();
            log.info("WebSocket 服务启动成功，端口={}", WS_PORT);
            channel.closeFuture().sync();
        } catch (InterruptedException e) {
   
            Thread.currentThread().interrupt();                 // 保留中断标记
            log.error("WebSocket 启动被中断", e);
        } catch (Exception e) {
   
            log.error("WebSocket 启动异常", e);
        } finally {
   
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
            log.info("WebSocket 服务已关闭");
        }
    }
}

4.2 独立端口是刻意的取舍

WebSocket 跑在 27051，与服务的 HTTP 端口完全分离。好处是互不干扰（HTTP 接口扩容不影响长连接）、可以独立调参；代价是必须处理端口映射与地址发现——这正是上一节粘性路由里要把注册端口替换成 WS_PORT 的原因。这是"独立部署"必然付出的复杂度，属于知情选择。

4.3 Pipeline 顺序不能错

@Override
protected void initChannel(SocketChannel ch) {
   
    ch.pipeline()
      .addLast(new HttpServerCodec())                      // HTTP 编解码（WS 握手基于 HTTP）
      .addLast(new ChunkedWriteHandler())                  // 支持大块数据分片写
      .addLast(new HttpObjectAggregator(65536))            // 聚合 HTTP 分片为完整请求（握手需要）
      .addLast(new WebSocketServerProtocolHandler("/ws/chat", true))  // 完成 101 协议升级
      .addLast(webSocketHandler);                          // 业务处理器
}

顺序的逻辑链条很清楚：WebSocket 握手本质是一次 HTTP 请求——先用 HTTP 编解码器，再聚合分片请求，才轮到协议升级处理器，最后才是业务 Handler。WebSocketServerProtocolHandler 的第二个参数是"是否处理 ping/pong 帧"，交给它处理可以省掉手写心跳应答。

4.4 单例 Handler 与线程安全

业务 Handler 用 @ChannelHandler.Sharable 声明为单例（这样才好统计在线数），配套要求是：所有共享状态必须线程安全，且不能用 @ChannelHandler 的实例字段存连接状态。

@Component
@ChannelHandler.Sharable                                   // 单例：便于全局在线统计
public class NioWebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
   

    /** 连接ID -> Channel */
    private static final Map<String, Channel> CHANNEL_MAP = new ConcurrentHashMap<>();
    /** 连接ID -> 绑定的账号/客户ID */
    private static final Map<String, Long> PRINCIPAL_MAP = new ConcurrentHashMap<>();

    /** 用 @Lazy 打破"启动期 Handler 依赖 Service、Service 又依赖容器"的循环 */
    @Lazy @Resource private ChannelConfigService channelConfigService;
    ...
}

五、连接生命周期：绑定、鉴权与清理

5.1 握手期完成鉴权与绑定

长连接只在握手时有一次"带凭证"的机会（业务消息里不该再传 token），所以鉴权必须在这里做完——正好复用上一篇的 JWT + Redis 双层鉴权：

@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
   
    if (evt instanceof WebSocketServerProtocolHandler.HandshakeComplete handshake) {
   
        MultiValueMap<String, String> params =
                UriComponentsBuilder.fromUriString(handshake.requestUri()).build().getQueryParams();
        String token = params.getFirst("token");
        if (StringUtils.isNotBlank(token)) {
   
            // 异步线程没有请求上下文，把鉴权头显式放进线程上下文，供 Feign 透传
            Map<String, String> headers = Map.of(
                    HttpHeaders.AUTHORIZATION, URLUtil.decode(token),
                    PLATFORM_HEADER, PlatformEnum.TENANT.name());
            ThreadHeaderHolder.setHeaders(headers);
            try {
   
                Result<LoginVO> result = authFeignClient.verifyToken();   // 走 JWT + Redis 双层校验
                if (result != null && result.getSuccess()) {
   
                    PRINCIPAL_MAP.put(ctx.channel().id().asLongText(),
                                      result.getData().getAccount().getId());
                }
            } finally {
   
                ThreadHeaderHolder.clear();     // 用完即清，避免污染 Netty 线程
            }
        }
    }
    ctx.fireUserEventTriggered(evt);
}

两个要点：① 未通过鉴权的连接不绑定身份，后续业务消息会被直接丢弃；② ThreadHeaderHolder 用完立刻清理——Netty 的 EventLoop 线程同样是长期复用的，ThreadLocal 残留会导致下一个连接读到别人的身份（与上一篇服务层拦截器的"双向清理"是同一类风险）。

5.2 防伪校验：别信客户端报的账号

客户端消息里带 accountId，但绝不能直接采信。对需要身份一致的通道，必须与握手期绑定的身份比对：

@Override
protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame frame) throws Exception {
   
    String key = ctx.channel().id().asLongText();
    ClientMessage message = objectMapper.readValue(frame.text(), ClientMessage.class);
    message.setChannelKey(key);

    // 管理端对话通道：消息里的 accountId 必须与连接绑定的身份一致，否则丢弃
    if (ChannelType.chat == message.getChannel()
            && !Objects.equals(PRINCIPAL_MAP.get(key), message.getAccountId())) {
   
        log.warn("身份不匹配，丢弃消息 key={} claimed={}", key, message.getAccountId());
        return;
    }

    switch (message.getType()) {
   
        case create        -> handleCreate(ctx, message);
        case heartbeat     -> sendHeartbeatAck(ctx, message);
        case msg           -> handleChatMessage(message);
        case clear         -> handleClear(message);
        default            -> log.warn("未处理的消息类型: {}", message.getType());
    }
}

5.3 三处清理，一处都不能漏

长连接最容易出的生产事故是内存泄漏——连接断了但映射表里的记录没删，跑几天内存就满了。所以"加入 / 移除 / 异常"三个入口都要成对清理：

@Override
public void handlerAdded(ChannelHandlerContext ctx) {
   
    CHANNEL_MAP.put(ctx.channel().id().asLongText(), ctx.channel());
    log.info("客户端接入 {}", ctx.channel().remoteAddress());
}

@Override
public void handlerRemoved(ChannelHandlerContext ctx) {
   
    String key = ctx.channel().id().asLongText();
    ctx.close();
    CHANNEL_MAP.remove(key);        // 两张表必须同步清理
    PRINCIPAL_MAP.remove(key);
    log.info("客户端断开 {}", ctx.channel().remoteAddress());
}

@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
   
    String key = ctx.channel().id().asLongText();
    log.error("连接异常 {}", ctx.channel().remoteAddress(), cause);
    CHANNEL_MAP.remove(key);        // 异常路径也要清，否则一定泄漏
    PRINCIPAL_MAP.remove(key);
    ctx.close();
}

另外注意 @ChannelHandler.Sharable 单例是全局共享的，所以 handlerAdded / handlerRemoved 会并发触发——映射表必须用 ConcurrentHashMap（这点我们踩过：早期用 HashMap 曾出现并发写导致条目丢失）。

六、消息协议：一个枚举搞定双向通信

客户端与服务端用同一套消息信封，靠 type 分发：

/** 双向消息类型 */
public enum MessageType {
   
    create,         // 建立会话（携带客户端配置）
    heartbeat,      // 心跳保活 / 心跳应答
    msg,            // 普通消息（含用户发送与机器人回复）
    historyList,    // 历史消息列表
    streamMsg,      // 流式分片（打字机效果）
    conversation,   // 会话列表变更（管理端）
    clear,          // 清除会话
}

/** 客户端 -> 服务端 */
@Data
public class ClientMessage {
   
    private String channelKey;      // 连接标识（服务端回填，客户端无需传）
    private ChannelType channel;    // web=网页客服，chat=管理端对话
    private MessageType type;
    private Long accountId;
    private JsonNode data;          // 按 type 多态解析
}

设计要点：data 用 JsonNode 承载多态载荷，按 type 再转换成具体 DTO——避免为一个枚举值建一个大而全的字段集合；channelKey 由服务端回填，客户端伪造无效（服务端以连接 ID 为准）。

七、踩坑与经验清单

粘性路由必须卡在 10000~10150 之间：早于 RouteToRequestUrlFilter 拿不到 lb:// URL，晚于 ReactiveLoadBalancerClientFilter 就已经被负载均衡改写过了；
WS 独立端口要同步改地址发现：服务注册的是 HTTP 端口，粘性路由必须替换成 WS_PORT，否则转发到 HTTP 端口上握手必失败；
多线程写同一连接必然可能乱序：只要有"流式 + 顺序敏感"的组合，就必须串行化写入（单线程执行器是最小代价解法）；
ThreadLocal 在 EventLoop 线程上同样会残留：Netty 线程也是长期复用的，用完必须 clear()；
客户端上报的身份一律不可信：必须与握手期绑定身份比对（越权风险）；
清理要覆盖三条路径：连接加入、正常断开、异常断开，两处映射表同步清理；
@Sharable 单例下所有静态状态都要线程安全：容器必须是并发容器；
异步启动 + 优雅关闭成对出现：启动不阻塞 Spring，关闭时 shutdownGracefully() 释放线程组。

八、结语

WebSocket 在多实例部署下的复杂度，本质来自两个"反直觉"：连接是有状态的（所以不能随便负载均衡）、写入是并发安全的但不保证全序（所以流式必须串行化）。

我们的应对可以浓缩成两句话：

网关负责"把连接粘对地方"——按客户端标识做粘性路由，跳过默认负载均衡；
服务负责"把消息按序送达"——Netty 原生独立服务 + 单线程写入队列保序。

再叠加握手期鉴权、身份防伪、三路清理这三件事，长连接才真正能扛住生产环境的并发与长时间运行。

当 WebSocket 遇上多实例部署：Netty 原生长连接与网关粘性路由实战复盘

一、背景：长连接在分布式下会遇到什么

二、难题一：网关粘性路由

2.1 为什么不能用默认负载均衡

2.2 关键在于过滤器顺序

2.3 两个必须知道的局限

三、难题二：流式消息为什么必须保序

3.1 根因在 EventLoop 的 MPSC 队列

3.2 解法：单线程串行化写入

四、Netty 服务搭建的四个工程细节

4.1 异步启动，不拖慢应用启动

4.2 独立端口是刻意的取舍

4.3 Pipeline 顺序不能错

4.4 单例 Handler 与线程安全

五、连接生命周期：绑定、鉴权与清理

5.1 握手期完成鉴权与绑定

5.2 防伪校验：别信客户端报的账号

5.3 三处清理，一处都不能漏

六、消息协议：一个枚举搞定双向通信

七、踩坑与经验清单

八、结语

热门文章

最新文章

相关电子书

探索云世界

热门

云计算

大数据

云原生

人工智能

数据库

开发与运维

活动广场

任务中心

训练营

直播

乘风者计划

下载

镜像站

技术资料

当 WebSocket 遇上多实例部署：Netty 原生长连接与网关粘性路由实战复盘

一、背景：长连接在分布式下会遇到什么

二、难题一：网关粘性路由

2.1 为什么不能用默认负载均衡

2.2 关键在于过滤器顺序

2.3 两个必须知道的局限

三、难题二：流式消息为什么必须保序

3.1 根因在 EventLoop 的 MPSC 队列

3.2 解法：单线程串行化写入

四、Netty 服务搭建的四个工程细节

4.1 异步启动，不拖慢应用启动

4.2 独立端口是刻意的取舍

4.3 Pipeline 顺序不能错

4.4 单例 Handler 与线程安全

五、连接生命周期：绑定、鉴权与清理

5.1 握手期完成鉴权与绑定

5.2 防伪校验：别信客户端报的账号

5.3 三处清理，一处都不能漏

六、消息协议：一个枚举搞定双向通信

七、踩坑与经验清单

八、结语

热门文章

最新文章

相关电子书