跳至主要內容

Netty 实战——自定义协议、心跳保活、连接管理完整方案

老张2026/4/30大约 8 分钟

Netty 实战——自定义协议、心跳保活、连接管理完整方案

适读人群：正在用 Netty 开发或准备开发网络服务的 Java 工程师 | 阅读时长：约 18 分钟 | 核心价值：给出生产可用的自定义协议设计、心跳机制和连接管理完整实现

光看 Netty 的 Echo 示例是不够的。真实项目里，你要面对三个必须解决的问题：

  1. 自定义协议：怎么定义消息格式，怎么编解码
  2. 心跳保活：连接空闲了怎么检测，怎么处理
  3. 连接管理：连接断了怎么重连，服务端怎么管理大量连接

这三个问题我在实际项目里都踩过坑，这篇文章给出完整的、能跑的解决方案。

自定义协议设计

我选的协议格式是长度前缀 + 消息头 + 消息体，这是我见过的实际项目里最常用的格式：

+---------+----------+--------+----------+----------+---------+
| magic   | version  | type   | serialId | bodyLen  | body    |
| 2 bytes | 1 byte   | 1 byte | 4 bytes  | 4 bytes  | N bytes |
+---------+----------+--------+----------+----------+---------+
  • magic：魔数，用来识别合法的数据包（防止垃圾数据）
  • version：协议版本，支持未来升级
  • type：消息类型（请求/响应/心跳）
  • serialId：消息序号，用来做请求-响应匹配
  • bodyLen：消息体长度
  • body：实际消息内容（用 JSON 或 Protobuf 序列化）
package com.example.netty.protocol;

/**
 * 自定义协议消息
 */
public class Message {

    public static final short MAGIC = (short) 0xCAFE;
    public static final byte VERSION = 1;

    // 消息类型常量
    public static final byte TYPE_REQUEST   = 0x01;
    public static final byte TYPE_RESPONSE  = 0x02;
    public static final byte TYPE_HEARTBEAT = 0x03;

    private short magic;
    private byte version;
    private byte type;
    private int serialId;
    private int bodyLen;
    private byte[] body;

    // 工厂方法：创建心跳消息
    public static Message heartbeat() {
        Message msg = new Message();
        msg.magic = MAGIC;
        msg.version = VERSION;
        msg.type = TYPE_HEARTBEAT;
        msg.serialId = 0;
        msg.bodyLen = 0;
        msg.body = new byte[0];
        return msg;
    }

    // Getter/Setter 省略...
    public short getMagic() { return magic; }
    public byte getType() { return type; }
    public int getSerialId() { return serialId; }
    public byte[] getBody() { return body; }
    public void setMagic(short magic) { this.magic = magic; }
    public void setVersion(byte version) { this.version = version; }
    public void setType(byte type) { this.type = type; }
    public void setSerialId(int serialId) { this.serialId = serialId; }
    public void setBodyLen(int bodyLen) { this.bodyLen = bodyLen; }
    public void setBody(byte[] body) { this.body = body; this.bodyLen = body.length; }
}

编码器（出站方向，把 Message 转成字节）：

package com.example.netty.protocol;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;

/**
 * 消息编码器：Message 对象 -> 字节流
 */
public class MessageEncoder extends MessageToByteEncoder<Message> {

    @Override
    protected void encode(ChannelHandlerContext ctx, Message msg, ByteBuf out) {
        // 写入固定头部
        out.writeShort(msg.getMagic());       // 2 字节魔数
        out.writeByte(Message.VERSION);       // 1 字节版本
        out.writeByte(msg.getType());         // 1 字节消息类型
        out.writeInt(msg.getSerialId());      // 4 字节序号
        out.writeInt(msg.getBody().length);   // 4 字节消息体长度
        // 写入消息体
        if (msg.getBody().length > 0) {
            out.writeBytes(msg.getBody());
        }
    }
}

解码器（入站方向，把字节流转成 Message）：

package com.example.netty.protocol;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import java.util.List;

/**
 * 消息解码器：字节流 -> Message 对象
 * 继承 ByteToMessageDecoder，它会自动处理粘包/半包
 */
public class MessageDecoder extends ByteToMessageDecoder {

    // 固定头部长度 = 2(magic) + 1(version) + 1(type) + 4(serialId) + 4(bodyLen) = 12 字节
    private static final int HEADER_LENGTH = 12;

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
        // 如果可读字节不够一个完整头部，等待更多数据
        if (in.readableBytes() < HEADER_LENGTH) return;

        // 标记当前读位置，如果数据不够，回退
        in.markReaderIndex();

        // 读取并验证魔数
        short magic = in.readShort();
        if (magic != Message.MAGIC) {
            // 非法数据，关闭连接
            ctx.close();
            return;
        }

        byte version = in.readByte();
        byte type = in.readByte();
        int serialId = in.readInt();
        int bodyLen = in.readInt();

        // 检查消息体是否完整到达
        if (in.readableBytes() < bodyLen) {
            // 数据不够，回退读指针，等待更多数据
            in.resetReaderIndex();
            return;
        }

        // 数据完整，读取消息体
        byte[] body = new byte[bodyLen];
        if (bodyLen > 0) {
            in.readBytes(body);
        }

        // 构建 Message 对象，加入输出列表
        Message msg = new Message();
        msg.setMagic(magic);
        msg.setVersion(version);
        msg.setType(type);
        msg.setSerialId(serialId);
        msg.setBody(body);
        out.add(msg);
    }
}

心跳保活：IdleStateHandler + 自定义处理器

TCP 连接在长时间没有数据传输时，中间的防火墙、NAT 设备可能会悄悄断开连接，但客户端和服务端都不知道（这叫"半开连接"）。心跳的作用是定期发送小数据包，一是维持连接活跃，二是检测连接是否真的断了。

Netty 提供了 IdleStateHandler 来检测空闲状态：

package com.example.netty.heartbeat;

import io.netty.channel.*;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import com.example.netty.protocol.Message;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 客户端心跳 Handler
 * 策略：客户端负责发心跳，服务端负责检测
 */
public class ClientHeartbeatHandler extends ChannelInboundHandlerAdapter {

    // 最大允许的心跳超时次数，超过这个次数就认为连接断了
    private static final int MAX_IDLE_COUNT = 3;
    private final AtomicInteger idleCount = new AtomicInteger(0);

    /**
     * 在 Pipeline 里这样使用：
     * pipeline.addLast(new IdleStateHandler(0, 5, 0, TimeUnit.SECONDS));
     * // 0=读空闲检测秒数, 5=写空闲检测秒数(5秒没写就触发), 0=读写空闲检测秒数
     * pipeline.addLast(new ClientHeartbeatHandler());
     */
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) evt;
            if (event.state() == IdleState.WRITER_IDLE) {
                // 5秒没有发送数据，发一个心跳
                int count = idleCount.incrementAndGet();
                if (count > MAX_IDLE_COUNT) {
                    // 连续超时次数过多，主动关闭连接，触发重连
                    System.out.println("心跳超时，关闭连接: " + ctx.channel().remoteAddress());
                    ctx.close();
                } else {
                    System.out.println("发送心跳包 (count=" + count + ")");
                    ctx.writeAndFlush(Message.heartbeat());
                }
            }
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        // 收到任何消息（包括心跳响应），重置空闲计数
        idleCount.set(0);
        ctx.fireChannelRead(msg);
    }
}

服务端心跳检测 Handler：

package com.example.netty.heartbeat;

import io.netty.channel.*;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import com.example.netty.protocol.Message;

/**
 * 服务端：检测客户端是否还活着
 * 如果超过指定时间没收到任何数据（包括心跳），关闭连接
 */
public class ServerHeartbeatHandler extends ChannelInboundHandlerAdapter {

    /**
     * 在 Pipeline 里这样使用：
     * pipeline.addLast(new IdleStateHandler(15, 0, 0, TimeUnit.SECONDS));
     * // 15秒没收到数据就触发 READER_IDLE
     * pipeline.addLast(new ServerHeartbeatHandler());
     */
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) evt;
            if (event.state() == IdleState.READER_IDLE) {
                System.out.println("超过 15 秒没收到数据，关闭连接: " + ctx.channel().remoteAddress());
                ctx.close(); // 关闭连接
            }
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        if (msg instanceof Message) {
            Message message = (Message) msg;
            if (message.getType() == Message.TYPE_HEARTBEAT) {
                // 收到心跳，回一个心跳响应就行，不需要传给下一个 Handler
                ctx.writeAndFlush(Message.heartbeat());
                return;
            }
        }
        // 非心跳消息，传给下一个 Handler 处理
        ctx.fireChannelRead(msg);
    }
}

踩坑实录一：心跳时间设置不合理，连接被中间设备断了但自己不知道

我们有一个长连接服务，客户端设置的心跳间隔是 60 秒。上线后发现，某些用户的连接经常莫名其妙断掉，但没有任何异常日志。

排查后发现：这些用户走的是某个运营商的网关，这个网关对超过 30 秒没有数据的 TCP 连接会强制断开，但 TCP 层面的断开不会发 RST 包，只是悄悄把状态清除了。

服务端和客户端都以为连接还在，但实际上数据发不出去。服务端发数据才会收到 RST，但没有数据发时，连接在服务端看起来永远是正常的。

解法：心跳间隔不要超过 30 秒，建议设置为 15-20 秒。

连接管理：客户端重连

客户端连接断开后需要自动重连，这是长连接服务的标配：

package com.example.netty.reconnect;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import com.example.netty.protocol.*;
import com.example.netty.heartbeat.ClientHeartbeatHandler;
import java.util.concurrent.TimeUnit;

/**
 * 带自动重连的 Netty 客户端
 */
public class ReconnectableClient {

    private final String host;
    private final int port;
    private final NioEventLoopGroup group;
    private final Bootstrap bootstrap;
    private volatile Channel channel;

    // 重连次数，用于指数退避
    private volatile int reconnectAttempts = 0;
    private static final int MAX_RECONNECT_DELAY_SECONDS = 60;

    public ReconnectableClient(String host, int port) {
        this.host = host;
        this.port = port;
        this.group = new NioEventLoopGroup();
        this.bootstrap = new Bootstrap()
            .group(group)
            .channel(NioSocketChannel.class)
            .option(ChannelOption.TCP_NODELAY, true)
            .option(ChannelOption.SO_KEEPALIVE, true)
            .handler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) {
                    ch.pipeline().addLast(new MessageDecoder());
                    ch.pipeline().addLast(new MessageEncoder());
                    ch.pipeline().addLast(new IdleStateHandler(0, 5, 0, TimeUnit.SECONDS));
                    ch.pipeline().addLast(new ClientHeartbeatHandler());
                    ch.pipeline().addLast(new ReconnectHandler(ReconnectableClient.this));
                }
            });
    }

    public void connect() {
        bootstrap.connect(host, port).addListener((ChannelFutureListener) future -> {
            if (future.isSuccess()) {
                channel = future.channel();
                reconnectAttempts = 0; // 连接成功，重置重试计数
                System.out.println("连接成功: " + host + ":" + port);
            } else {
                scheduleReconnect();
            }
        });
    }

    void scheduleReconnect() {
        reconnectAttempts++;
        // 指数退避：1s, 2s, 4s, 8s ... 最大 60s
        long delay = Math.min(
            (long) Math.pow(2, reconnectAttempts - 1),
            MAX_RECONNECT_DELAY_SECONDS
        );
        System.out.println("连接失败，" + delay + " 秒后重试 (第 " + reconnectAttempts + " 次)");
        group.schedule(this::connect, delay, TimeUnit.SECONDS);
    }

    // 连接断开时触发重连的 Handler
    static class ReconnectHandler extends ChannelInboundHandlerAdapter {
        private final ReconnectableClient client;

        ReconnectHandler(ReconnectableClient client) {
            this.client = client;
        }

        @Override
        public void channelInactive(ChannelHandlerContext ctx) {
            System.out.println("连接断开，准备重连...");
            client.scheduleReconnect();
        }
    }
}

踩坑实录二：重连时没有指数退避，把服务端打垮了

我早期实现的重连逻辑是：连接断了，等 1 秒，重连；还失败，等 1 秒，重连；如此循环。

某次服务端重启时，几千个客户端同时断线，1 秒后同时重连，服务端瞬间压力暴增，直接过载，反而连正常启动都受影响。

必须用指数退避 + 随机抖动：

// 加随机抖动，防止所有客户端同时重连（惊群效应）
long baseDelay = (long) Math.pow(2, reconnectAttempts - 1);
long jitter = (long) (Math.random() * baseDelay * 0.3); // 加 30% 随机抖动
long delay = Math.min(baseDelay + jitter, MAX_RECONNECT_DELAY_SECONDS);

踩坑实录三：ChannelGroup 管理连接，忘了处理连接断开时的清理

服务端要管理大量连接，通常用 ChannelGroup：

package com.example.netty.management;

import io.netty.channel.*;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;

/**
 * 服务端连接管理
 * ChannelGroup 会在 Channel 关闭时自动移除，不需要手动清理
 */
@ChannelHandler.Sharable
public class ConnectionManager extends ChannelInboundHandlerAdapter {

    // DefaultChannelGroup 是线程安全的，可以 @Sharable
    private static final ChannelGroup ALL_CHANNELS =
        new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        ALL_CHANNELS.add(ctx.channel()); // 新连接加入
        System.out.println("当前连接数: " + ALL_CHANNELS.size());
        ctx.fireChannelActive();
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        // ChannelGroup 会自动处理：Channel 关闭时从 Group 中移除
        // 不需要手动 ALL_CHANNELS.remove(ctx.channel())
        System.out.println("连接断开，当前连接数: " + ALL_CHANNELS.size());
        ctx.fireChannelInactive();
    }

    // 广播消息给所有连接
    public static void broadcast(Object message) {
        ALL_CHANNELS.writeAndFlush(message);
    }

    public static int getConnectionCount() {
        return ALL_CHANNELS.size();
    }
}

DefaultChannelGroup 的一个好处是：Channel 关闭时会自动从 Group 里移除，不需要手动清理。我之前用 ConcurrentHashMap 手动维护连接映射，结果连接断开后忘了清理，Map 里堆积了大量已断开的 Channel，内存一直涨。

这三个方面——协议设计、心跳保活、连接管理——是 Netty 实际项目中必须处理好的基础。代码可以直接参考，但要根据具体业务调整参数（心跳间隔、重连延迟、最大连接数等）。

编辑此页
最近更新:
贡献者: xiaozhangge