Netty教程 / 第 70 节

第7章：编解码器（Codec）

本章导读

编解码器是 Netty 处理网络数据的核心组件。本章将讲解为什么需要编解码器、Netty 内置的编解码器、如何自定义编解码器，以及如何解决粘包/拆包问题。

7.1 为什么需要编解码器

7.1.1 网络传输的问题

问题：网络传输只能传输字节流

应用层对象 → 字节流 → 网络传输 → 字节流 → 应用层对象
          编码                      解码

示例：

// 发送端：需要将对象转换为字节
User user = new User("Alice", 25);
byte[] bytes = serialize(user);  // 编码
channel.writeAndFlush(bytes);

// 接收端：需要将字节转换为对象
byte[] bytes = receive();
User user = deserialize(bytes);  // 解码

7.1.2 编解码器的作用

编码器（Encoder）：

将消息对象转换为字节流
出站操作（写入数据时）

解码器（Decoder）：

将字节流转换为消息对象
入站操作（读取数据时）

编解码器（Codec）：

同时包含编码和解码功能

7.2 内置编解码器

7.2.1 字符串编解码器

pipeline.addLast("decoder", new StringDecoder(CharsetUtil.UTF_8));
pipeline.addLast("encoder", new StringEncoder(CharsetUtil.UTF_8));

// 使用
public class StringHandler extends SimpleChannelInboundHandler<String> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) {
        System.out.println("收到字符串: " + msg);
        ctx.writeAndFlush("回复: " + msg);  // 自动编码为字节
    }
}

7.2.2 对象编解码器

// Java 序列化（不推荐，性能差）
pipeline.addLast(new ObjectDecoder(ClassResolvers.cacheDisabled(null)));
pipeline.addLast(new ObjectEncoder());

// 使用
public class ObjectHandler extends SimpleChannelInboundHandler<Serializable> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Serializable msg) {
        System.out.println("收到对象: " + msg);
    }
}

7.2.3 Base64 编解码器

pipeline.addLast(new Base64Decoder());
pipeline.addLast(new Base64Encoder());

7.2.4 压缩编解码器

// Gzip 压缩
pipeline.addLast(new JdkZlibEncoder(ZlibWrapper.GZIP));
pipeline.addLast(new JdkZlibDecoder(ZlibWrapper.GZIP));

// Snappy 压缩
pipeline.addLast(new SnappyFrameEncoder());
pipeline.addLast(new SnappyFrameDecoder());

7.3 自定义编解码器

7.3.1 MessageToByteEncoder

定义：将消息对象编码为字节流

public class CustomMessageEncoder extends MessageToByteEncoder<CustomMessage> {
    
    @Override
    protected void encode(ChannelHandlerContext ctx, CustomMessage msg, ByteBuf out) {
        // 编码逻辑
        out.writeInt(msg.getId());
        
        byte[] nameBytes = msg.getName().getBytes(CharsetUtil.UTF_8);
        out.writeInt(nameBytes.length);
        out.writeBytes(nameBytes);
        
        byte[] contentBytes = msg.getContent().getBytes(CharsetUtil.UTF_8);
        out.writeInt(contentBytes.length);
        out.writeBytes(contentBytes);
    }
}

完整示例：

/**
 * 自定义消息格式：
 * +------+-------------+------+---------+------+---------+
 * |  ID  | Name Length | Name | Content | Len  | Content |
 * | 4字节|   4字节     | N字节|  Length | 4字节| N字节   |
 * +------+-------------+------+---------+------+---------+
 */
public class Message {
    private int id;
    private String name;
    private String content;
    
    // getter/setter...
}

public class MessageEncoder extends MessageToByteEncoder<Message> {
    
    @Override
    protected void encode(ChannelHandlerContext ctx, Message msg, ByteBuf out) {
        // 1. 写入 ID
        out.writeInt(msg.getId());
        
        // 2. 写入 Name
        byte[] nameBytes = msg.getName().getBytes(CharsetUtil.UTF_8);
        out.writeInt(nameBytes.length);
        out.writeBytes(nameBytes);
        
        // 3. 写入 Content
        byte[] contentBytes = msg.getContent().getBytes(CharsetUtil.UTF_8);
        out.writeInt(contentBytes.length);
        out.writeBytes(contentBytes);
    }
}

7.3.2 ByteToMessageDecoder

定义：将字节流解码为消息对象

public class MessageDecoder extends ByteToMessageDecoder {
    
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
        // 至少需要 12 字节（ID + Name Length + Content Length）
        if (in.readableBytes() < 12) {
            return;
        }
        
        // 标记读取位置
        in.markReaderIndex();
        
        // 1. 读取 ID
        int id = in.readInt();
        
        // 2. 读取 Name
        int nameLength = in.readInt();
        if (in.readableBytes() < nameLength + 4) {
            in.resetReaderIndex();
            return;
        }
        byte[] nameBytes = new byte[nameLength];
        in.readBytes(nameBytes);
        String name = new String(nameBytes, CharsetUtil.UTF_8);
        
        // 3. 读取 Content
        int contentLength = in.readInt();
        if (in.readableBytes() < contentLength) {
            in.resetReaderIndex();
            return;
        }
        byte[] contentBytes = new byte[contentLength];
        in.readBytes(contentBytes);
        String content = new String(contentBytes, CharsetUtil.UTF_8);
        
        // 4. 创建消息对象
        Message msg = new Message();
        msg.setId(id);
        msg.setName(name);
        msg.setContent(content);
        
        out.add(msg);
    }
}

7.3.3 MessageToMessageEncoder

定义：将一种消息类型转换为另一种消息类型

public class IntegerToStringEncoder extends MessageToMessageEncoder<Integer> {
    
    @Override
    protected void encode(ChannelHandlerContext ctx, Integer msg, List<Object> out) {
        out.add(String.valueOf(msg));
    }
}

7.3.4 MessageToMessageDecoder

public class StringToIntegerDecoder extends MessageToMessageDecoder<String> {
    
    @Override
    protected void decode(ChannelHandlerContext ctx, String msg, List<Object> out) {
        out.add(Integer.parseInt(msg));
    }
}

7.3.5 ByteToMessageCodec

定义：同时包含编码和解码功能

public class MessageCodec extends ByteToMessageCodec<Message> {
    
    @Override
    protected void encode(ChannelHandlerContext ctx, Message msg, ByteBuf out) {
        // 编码逻辑（同 MessageEncoder）
        out.writeInt(msg.getId());
        // ...
    }
    
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
        // 解码逻辑（同 MessageDecoder）
        if (in.readableBytes() < 12) {
            return;
        }
        // ...
    }
}

7.4 解决粘包/拆包问题

7.4.1 什么是粘包/拆包

粘包：多个数据包粘在一起

发送: [包1] [包2] [包3]
接收: [包1包2包3]

拆包：一个数据包被拆分

发送: [大包]
接收: [大包的前半部分] [大包的后半部分]

原因：

TCP 是流式协议，没有消息边界
Nagle 算法合并小包
MSS 限制导致拆包

7.4.2 解决方案

方案1：固定长度

// 每个消息固定 100 字节
pipeline.addLast(new FixedLengthFrameDecoder(100));

方案2：分隔符

// 使用换行符分隔
pipeline.addLast(new LineBasedFrameDecoder(1024));

// 使用自定义分隔符
ByteBuf delimiter = Unpooled.copiedBuffer("$$".getBytes());
pipeline.addLast(new DelimiterBasedFrameDecoder(1024, delimiter));

方案3：长度字段（推荐）

/**
 * 消息格式：
 * +--------+----------+
 * | Length | Content  |
 * | 4 byte | N bytes  |
 * +--------+----------+
 */
pipeline.addLast(new LengthFieldBasedFrameDecoder(
    1024,    // maxFrameLength: 最大帧长度
    0,       // lengthFieldOffset: 长度字段偏移量
    4,       // lengthFieldLength: 长度字段长度
    0,       // lengthAdjustment: 长度调整值
    4        // initialBytesToStrip: 跳过的字节数
));

pipeline.addLast(new LengthFieldPrepender(4));  // 编码器

方案4：自定义协议

public class CustomFrameDecoder extends ByteToMessageDecoder {
    
    private static final int HEADER_SIZE = 8;  // 魔数(4) + 长度(4)
    private static final int MAGIC_NUMBER = 0x12345678;
    
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
        // 至少需要头部
        if (in.readableBytes() < HEADER_SIZE) {
            return;
        }
        
        in.markReaderIndex();
        
        // 读取魔数
        int magic = in.readInt();
        if (magic != MAGIC_NUMBER) {
            throw new RuntimeException("非法的魔数: " + magic);
        }
        
        // 读取长度
        int length = in.readInt();
        
        // 检查数据是否完整
        if (in.readableBytes() < length) {
            in.resetReaderIndex();
            return;
        }
        
        // 读取数据
        byte[] data = new byte[length];
        in.readBytes(data);
        
        out.add(data);
    }
}

7.5 常用编解码器详解

7.5.1 LengthFieldBasedFrameDecoder

参数说明：

new LengthFieldBasedFrameDecoder(
    maxFrameLength,        // 最大帧长度
    lengthFieldOffset,     // 长度字段偏移量
    lengthFieldLength,     // 长度字段长度（1/2/3/4/8字节）
    lengthAdjustment,      // 长度调整值
    initialBytesToStrip    // 跳过的字节数
);

示例1：长度字段在开头

消息格式：
+--------+----------+
| Length | Content  |
| 2 byte | N bytes  |
+--------+----------+

配置：
new LengthFieldBasedFrameDecoder(1024, 0, 2, 0, 2);

示例2：长度字段在中间

消息格式：
+------+--------+----------+
| Type | Length | Content  |
| 1 B  | 2 B    | N bytes  |
+------+--------+----------+

配置：
new LengthFieldBasedFrameDecoder(1024, 1, 2, 0, 3);

示例3：长度包含头部

消息格式：
+--------+----------+
| Length | Content  |  Length = Content长度 + 4
| 4 B    | N bytes  |
+--------+----------+

配置：
new LengthFieldBasedFrameDecoder(1024, 0, 4, -4, 4);

7.5.2 完整示例

public class CodecServer {
    
    public static void main(String[] args) throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        
        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) {
                            ChannelPipeline pipeline = ch.pipeline();
                            
                            // 1. 解决粘包/拆包
                            pipeline.addLast(new LengthFieldBasedFrameDecoder(
                                    1024, 0, 4, 0, 4));
                            pipeline.addLast(new LengthFieldPrepender(4));
                            
                            // 2. 自定义编解码器
                            pipeline.addLast(new MessageDecoder());
                            pipeline.addLast(new MessageEncoder());
                            
                            // 3. 业务Handler
                            pipeline.addLast(new MessageHandler());
                        }
                    });
            
            ChannelFuture future = bootstrap.bind(8080).sync();
            System.out.println("服务器启动成功");
            future.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

class MessageHandler extends SimpleChannelInboundHandler<Message> {
    
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Message msg) {
        System.out.println("收到消息: " + msg);
        
        // 回复
        Message response = new Message();
        response.setId(msg.getId());
        response.setName("Server");
        response.setContent("收到: " + msg.getContent());
        
        ctx.writeAndFlush(response);
    }
}

7.6 Protocol Buffers 集成

7.6.1 定义 .proto 文件

syntax = "proto3";

option java_package = "com.netty.tutorial.proto";
option java_outer_classname = "MessageProto";

message Message {
    int32 id = 1;
    string name = 2;
    string content = 3;
}

7.6.2 使用 Protobuf 编解码器

pipeline.addLast(new ProtobufVarint32FrameDecoder());
pipeline.addLast(new ProtobufDecoder(MessageProto.Message.getDefaultInstance()));
pipeline.addLast(new ProtobufVarint32LengthFieldPrepender());
pipeline.addLast(new ProtobufEncoder());

// 业务Handler
public class ProtobufHandler extends SimpleChannelInboundHandler<MessageProto.Message> {
    
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, MessageProto.Message msg) {
        System.out.println("收到消息: " + msg.getContent());
        
        // 构建响应
        MessageProto.Message response = MessageProto.Message.newBuilder()
                .setId(msg.getId())
                .setName("Server")
                .setContent("收到: " + msg.getContent())
                .build();
        
        ctx.writeAndFlush(response);
    }
}

7.7 本章小结

本章我们学习了编解码器：

✅ 编解码器的作用：对象与字节流的转换
✅ 内置编解码器：String、Object、Base64、压缩等
✅ 自定义编解码器：MessageToByteEncoder、ByteToMessageDecoder
✅ 粘包/拆包：固定长度、分隔符、长度字段、自定义协议
✅ LengthFieldBasedFrameDecoder：最常用的解决方案
✅ Protocol Buffers：高效的序列化方案

关键要点

编码器是出站操作，解码器是入站操作
粘包/拆包是 TCP 的固有问题，需要应用层解决
LengthFieldBasedFrameDecoder 是最常用的解决方案
Protocol Buffers 性能优于 Java 序列化
自定义编解码器要处理好半包问题

下一章预告

下一章我们将深入学习粘包与拆包解决方案的更多细节和实战案例。

练习题

基础题：实现一个简单的字符串编解码器
进阶题：使用 LengthFieldBasedFrameDecoder 实现自定义协议
挑战题：集成 Protocol Buffers 实现高性能通信

上一章：第6章：ByteBuf缓冲区
下一章：第8章：粘包与拆包解决方案