Netty编解码案例

wen java案例 2026-08-05 258

本文目录导读：

Netty编解码案例

基础项目结构
协议定义
编码器实现
解码器实现
组合编解码器
服务器端实现
客户端实现
优化版本（使用JSON编解码）
测试类
使用Protobuf的版本（高级）

我将为您提供一个完整的Netty编解码案例,涵盖常见的编解码器实现方式。

基础项目结构

// 项目结构
src/main/java/com/example/netty/
├── protocol/
│   ├── Message.java          // 消息对象
│   ├── MessageType.java      // 消息类型枚举
│   └── ProtocolConstants.java // 协议常量
├── codec/
│   ├── MessageEncoder.java   // 消息编码器
│   ├── MessageDecoder.java   // 消息解码器
│   └── MessageCodec.java     // 组合编解码器
├── server/
│   └── NettyServer.java      // 服务器端
└── client/
    └── NettyClient.java      // 客户端

协议定义

package com.example.netty.protocol;
// 消息类型枚举
public enum MessageType {
    LOGIN_REQUEST(1, "登录请求"),
    LOGIN_RESPONSE(2, "登录响应"),
    CHAT_REQUEST(3, "聊天请求"),
    CHAT_RESPONSE(4, "聊天响应"),
    HEARTBEAT(5, "心跳");
    private final int code;
    private final String description;
    MessageType(int code, String description) {
        this.code = code;
        this.description = description;
    }
    public int getCode() {
        return code;
    }
    public String getDescription() {
        return description;
    }
    public static MessageType getByCode(int code) {
        for (MessageType type : MessageType.values()) {
            if (type.code == code) {
                return type;
            }
        }
        return null;
    }
}

package com.example.netty.protocol;
import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;
// 消息对象
public class Message implements Serializable {
    private static final long serialVersionUID = 1L;
    // 消息头
    private int magicNumber;      // 魔数
    private byte version;         // 版本号
    private int messageType;      // 消息类型
    private int sequenceId;       // 序列号
    private int length;           // 消息体长度
    // 消息体
    private Map<String, Object> body = new HashMap<>();
    // 构造函数
    public Message() {}
    public Message(int magicNumber, byte version, int messageType, int sequenceId) {
        this.magicNumber = magicNumber;
        this.version = version;
        this.messageType = messageType;
        this.sequenceId = sequenceId;
    }
    // getter和setter方法
    public int getMagicNumber() { return magicNumber; }
    public void setMagicNumber(int magicNumber) { this.magicNumber = magicNumber; }
    public byte getVersion() { return version; }
    public void setVersion(byte version) { this.version = version; }
    public int getMessageType() { return messageType; }
    public void setMessageType(int messageType) { this.messageType = messageType; }
    public int getSequenceId() { return sequenceId; }
    public void setSequenceId(int sequenceId) { this.sequenceId = sequenceId; }
    public int getLength() { return length; }
    public void setLength(int length) { this.length = length; }
    public Map<String, Object> getBody() { return body; }
    public void setBody(Map<String, Object> body) { this.body = body; }
    // 便利方法
    public void put(String key, Object value) {
        body.put(key, value);
    }
    public Object get(String key) {
        return body.get(key);
    }
}

package com.example.netty.protocol;
// 协议常量
public class ProtocolConstants {
    public static final int MAGIC_NUMBER = 0xCAFEBABE;  // 魔数
    public static final byte VERSION = 1;                 // 版本号
    public static final int HEADER_LENGTH = 18;           // 头部长度 (4+1+4+4+4+1)
    // 内存分配相关
    public static final int INITIAL_CAPACITY = 256;
    public static final int MAX_FRAME_LENGTH = 1024 * 1024; // 1MB
}

编码器实现

package com.example.netty.codec;
import com.example.netty.protocol.Message;
import com.example.netty.protocol.ProtocolConstants;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;
import java.nio.charset.StandardCharsets;
import java.util.Map;
// 消息编码器
public class MessageEncoder extends MessageToByteEncoder<Message> {
    @Override
    protected void encode(ChannelHandlerContext ctx, Message msg, ByteBuf out) throws Exception {
        // 写入魔数
        out.writeInt(ProtocolConstants.MAGIC_NUMBER);
        // 写入版本号
        out.writeByte(ProtocolConstants.VERSION);
        // 写入消息类型
        out.writeInt(msg.getMessageType());
        // 写入序列号
        out.writeInt(msg.getSequenceId());
        // 序列化消息体
        byte[] bodyBytes = serializeBody(msg.getBody());
        // 写入消息体长度
        out.writeInt(bodyBytes.length);
        // 写入校验位（简单示例，实际可用CRC32等）
        out.writeByte(calculateChecksum(bodyBytes));
        // 写入消息体
        out.writeBytes(bodyBytes);
    }
    // 序列化消息体（使用Java序列化）
    private byte[] serializeBody(Map<String, Object> body) throws Exception {
        if (body == null || body.isEmpty()) {
            return new byte[0];
        }
        // 简单实现：将Map转为JSON字符串
        // 实际项目中可以使用FastJSON、Jackson等库
        StringBuilder sb = new StringBuilder();
        sb.append('{');
        for (Map.Entry<String, Object> entry : body.entrySet()) {
            sb.append('"').append(entry.getKey()).append("\":\"")
              .append(entry.getValue()).append("\",");
        }
        if (sb.length() > 1) {
            sb.deleteCharAt(sb.length() - 1);
        }
        sb.append('}');
        return sb.toString().getBytes(StandardCharsets.UTF_8);
    }
    // 计算校验和（简单实现）
    private byte calculateChecksum(byte[] data) {
        byte checksum = 0;
        for (byte b : data) {
            checksum ^= b;
        }
        return checksum;
    }
}

解码器实现

package com.example.netty.codec;
import com.example.netty.protocol.Message;
import com.example.netty.protocol.ProtocolConstants;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
// 消息解码器
public class MessageDecoder extends ByteToMessageDecoder {
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        // 检查是否有足够的数据读取头部
        if (in.readableBytes() < ProtocolConstants.HEADER_LENGTH) {
            return; // 等待更多数据
        }
        // 标记当前读取位置
        in.markReaderIndex();
        // 读取魔数
        int magicNumber = in.readInt();
        if (magicNumber != ProtocolConstants.MAGIC_NUMBER) {
            // 魔数不匹配，可能是垃圾数据
            in.resetReaderIndex();
            throw new Exception("Invalid magic number: " + magicNumber);
        }
        // 读取版本号
        byte version = in.readByte();
        // 读取消息类型
        int messageType = in.readInt();
        // 读取序列号
        int sequenceId = in.readInt();
        // 读取消息体长度
        int length = in.readInt();
        // 检查消息体长度是否合理
        if (length < 0 || length > ProtocolConstants.MAX_FRAME_LENGTH) {
            throw new Exception("Invalid message length: " + length);
        }
        // 读取校验位
        byte checksum = in.readByte();
        // 检查是否有足够的数据读取消息体
        if (in.readableBytes() < length) {
            // 数据不完整，重置读取位置
            in.resetReaderIndex();
            return;
        }
        // 读取消息体
        byte[] bodyBytes = new byte[length];
        in.readBytes(bodyBytes);
        // 验证校验和
        if (checksum != calculateChecksum(bodyBytes)) {
            throw new Exception("Checksum verification failed");
        }
        // 构建消息对象
        Message message = new Message(magicNumber, version, messageType, sequenceId);
        message.setLength(length);
        // 反序列化消息体
        message.setBody(deserializeBody(bodyBytes));
        // 将解码后的消息加入输出列表
        out.add(message);
    }
    // 反序列化消息体
    private Map<String, Object> deserializeBody(byte[] data) {
        Map<String, Object> body = new HashMap<>();
        if (data.length == 0) {
            return body;
        }
        // 简单解析JSON格式（实际使用JSON库）
        String json = new String(data, StandardCharsets.UTF_8);
        json = json.substring(1, json.length() - 1); // 去掉花括号
        if (!json.isEmpty()) {
            String[] pairs = json.split(",");
            for (String pair : pairs) {
                String[] keyValue = pair.split(":");
                if (keyValue.length == 2) {
                    String key = keyValue[0].replace("\"", "");
                    String value = keyValue[1].replace("\"", "");
                    body.put(key, value);
                }
            }
        }
        return body;
    }
    // 计算校验和
    private byte calculateChecksum(byte[] data) {
        byte checksum = 0;
        for (byte b : data) {
            checksum ^= b;
        }
        return checksum;
    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        // 捕获解码异常
        System.err.println("Decode error: " + cause.getMessage());
        cause.printStackTrace();
    }
}

组合编解码器

package com.example.netty.codec;
import io.netty.channel.CombinedChannelDuplexHandler;
// 组合编解码器
public class MessageCodec extends CombinedChannelDuplexHandler<MessageDecoder, MessageEncoder> {
    public MessageCodec() {
        super(new MessageDecoder(), new MessageEncoder());
    }
}

服务器端实现

package com.example.netty.server;
import com.example.netty.codec.MessageCodec;
import com.example.netty.protocol.Message;
import com.example.netty.protocol.MessageType;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateHandler;
public class NettyServer {
    private int port;
    public NettyServer(int port) {
        this.port = port;
    }
    public void start() throws InterruptedException {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .option(ChannelOption.SO_BACKLOG, 128)
                .childOption(ChannelOption.SO_KEEPALIVE, true)
                .handler(new LoggingHandler(LogLevel.INFO))
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        // 添加空闲检测
                        pipeline.addLast(new IdleStateHandler(60, 0, 0));
                        // 添加编解码器
                        pipeline.addLast(new MessageCodec());
                        // 添加业务处理器
                        pipeline.addLast(new ServerHandler());
                    }
                });
            // 绑定端口，同步等待成功
            ChannelFuture future = bootstrap.bind(port).sync();
            System.out.println("Server started on port " + port);
            // 等待服务器关闭
            future.channel().closeFuture().sync();
        } finally {
            // 优雅关闭
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
    // 服务器处理器
    @ChannelHandler.Sharable
    public static class ServerHandler extends ChannelInboundHandlerAdapter {
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            System.out.println("Client connected: " + ctx.channel().remoteAddress());
        }
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            if (msg instanceof Message) {
                Message message = (Message) msg;
                System.out.println("Received message: " + message);
                // 处理消息
                Message response = processMessage(ctx.channel(), message);
                // 发送响应
                if (response != null) {
                    ctx.writeAndFlush(response);
                }
            }
        }
        private Message processMessage(Channel channel, Message message) {
            MessageType type = MessageType.getByCode(message.getMessageType());
            if (type == null) {
                return null;
            }
            switch (type) {
                case LOGIN_REQUEST:
                    // 处理登录请求
                    return handleLogin(message);
                case CHAT_REQUEST:
                    // 处理聊天请求
                    return handleChat(channel, message);
                case HEARTBEAT:
                    // 处理心跳
                    return handleHeartbeat(message);
                default:
                    System.out.println("Unsupported message type: " + message.getMessageType());
                    return null;
            }
        }
        private Message handleLogin(Message message) {
            Message response = createMessage(MessageType.LOGIN_RESPONSE, message.getSequenceId());
            String username = (String) message.get("username");
            String password = (String) message.get("password");
            // 简单验证
            if ("admin".equals(username) && "123456".equals(password)) {
                response.put("code", 200);
                response.put("message", "登录成功");
                System.out.println("User login success: " + username);
            } else {
                response.put("code", 401);
                response.put("message", "登录失败");
                System.out.println("User login failed: " + username);
            }
            return response;
        }
        private Message handleChat(Channel channel, Message message) {
            Message response = createMessage(MessageType.CHAT_RESPONSE, message.getSequenceId());
            String content = (String) message.get("content");
            System.out.println("Received chat message: " + content);
            response.put("code", 200);
            response.put("message", "消息已收到");
            response.put("echo", content);
            return response;
        }
        private Message handleHeartbeat(Message message) {
            Message response = createMessage(MessageType.HEARTBEAT, message.getSequenceId());
            response.put("status", 1);
            return response;
        }
        private Message createMessage(MessageType type, int sequenceId) {
            return new Message(
                0xCAFEBABE,
                (byte) 1,
                type.getCode(),
                sequenceId
            );
        }
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
            ctx.flush();
        }
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            System.err.println("Server exception: " + cause.getMessage());
            cause.printStackTrace();
            ctx.close();
        }
    }
    public static void main(String[] args) throws InterruptedException {
        new NettyServer(8080).start();
    }
}

客户端实现

package com.example.netty.client;
import com.example.netty.codec.MessageCodec;
import com.example.netty.protocol.Message;
import com.example.netty.protocol.MessageType;
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 java.util.concurrent.TimeUnit;
public class NettyClient {
    private String host;
    private int port;
    private Channel channel;
    public NettyClient(String host, int port) {
        this.host = host;
        this.port = port;
    }
    public void connect() throws InterruptedException {
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast(new IdleStateHandler(0, 30, 0));
                        pipeline.addLast(new MessageCodec());
                        pipeline.addLast(new ClientHandler());
                    }
                });
            // 连接服务器
            ChannelFuture future = bootstrap.connect(host, port).sync();
            channel = future.channel();
            System.out.println("Connected to server: " + host + ":" + port);
        } finally {
            // 注意：这里不关闭group，保持连接
        }
    }
    // 发送登录请求
    public void sendLogin(String username, String password) throws InterruptedException {
        Message message = createMessage(MessageType.LOGIN_REQUEST);
        message.put("username", username);
        message.put("password", password);
        if (channel != null && channel.isActive()) {
            channel.writeAndFlush(message).sync();
        }
    }
    // 发送聊天消息
    public void sendChat(String content) throws InterruptedException {
        Message message = createMessage(MessageType.CHAT_REQUEST);
        message.put("content", content);
        if (channel != null && channel.isActive()) {
            channel.writeAndFlush(message).sync();
        }
    }
    private Message createMessage(MessageType type) {
        // 生成一个简单的序列号
        int sequenceId = (int) (System.nanoTime() % 1000000);
        return new Message(0xCAFEBABE, (byte) 1, type.getCode(), sequenceId);
    }
    // 客户端处理器
    public static class ClientHandler extends ChannelInboundHandlerAdapter {
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            System.out.println("Connected to server");
        }
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            if (msg instanceof Message) {
                Message message = (Message) msg;
                System.out.println("Received response: " + message);
                // 处理响应
                MessageType type = MessageType.getByCode(message.getMessageType());
                if (type != null) {
                    switch (type) {
                        case LOGIN_RESPONSE:
                            handleLoginResponse(message);
                            break;
                        case CHAT_RESPONSE:
                            handleChatResponse(message);
                            break;
                        case HEARTBEAT:
                            handleHeartbeatResponse(message);
                            break;
                        default:
                            System.out.println("Unknown response type");
                    }
                }
            }
        }
        private void handleLoginResponse(Message message) {
            Integer code = (Integer) message.get("code");
            String msg = (String) message.get("message");
            System.out.println("Login response - Code: " + code + ", Message: " + msg);
        }
        private void handleChatResponse(Message message) {
            Integer code = (Integer) message.get("code");
            String msg = (String) message.get("message");
            String echo = (String) message.get("echo");
            System.out.println("Chat response - Code: " + code + ", Message: " + msg + ", Echo: " + echo);
        }
        private void handleHeartbeatResponse(Message message) {
            System.out.println("Heartbeat response received");
        }
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            System.err.println("Client exception: " + cause.getMessage());
            cause.printStackTrace();
            ctx.close();
        }
    }
    public static void main(String[] args) throws InterruptedException {
        NettyClient client = new NettyClient("localhost", 8080);
        client.connect();
        // 测试登录
        client.sendLogin("admin", "123456");
        Thread.sleep(2000);
        // 测试聊天
        client.sendChat("Hello Netty!");
        Thread.sleep(2000);
        // 保持连接一段时间
        TimeUnit.SECONDS.sleep(10);
    }
}

优化版本（使用JSON编解码）

package com.example.netty.codec;
import com.alibaba.fastjson.JSON;
import com.example.netty.protocol.Message;
import com.example.netty.protocol.ProtocolConstants;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.MessageToByteEncoder;
import io.netty.handler.codec.LengthFieldPrepender;
// 使用LengthFieldBasedFrameDecoder和LengthFieldPrepender的版本
public class OptimizedMessageEncoder extends MessageToByteEncoder<Message> {
    @Override
    protected void encode(ChannelHandlerContext ctx, Message msg, ByteBuf out) throws Exception {
        // 使用JSON序列化整个消息对象
        byte[] data = JSON.toJSONBytes(msg);
        out.writeBytes(data);
    }
}
// 优化后的解码器（使用LengthFieldBasedFrameDecoder处理粘包/拆包）
public class OptimizedMessageDecoder extends LengthFieldBasedFrameDecoder {
    private static final int MAX_FRAME_LENGTH = 1024 * 1024; // 1MB
    private static final int LENGTH_FIELD_OFFSET = 0;
    private static final int LENGTH_FIELD_LENGTH = 4;
    private static final int LENGTH_ADJUSTMENT = 0;
    private static final int INITIAL_BYTES_TO_STRIP = 4;
    public OptimizedMessageDecoder() {
        super(MAX_FRAME_LENGTH, 
              LENGTH_FIELD_OFFSET, 
              LENGTH_FIELD_LENGTH, 
              LENGTH_ADJUSTMENT,
              INITIAL_BYTES_TO_STRIP);
    }
    @Override
    protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
        ByteBuf frame = (ByteBuf) super.decode(ctx, in);
        if (frame == null) {
            return null;
        }
        try {
            byte[] data = new byte[frame.readableBytes()];
            frame.readBytes(data);
            return JSON.parseObject(data, Message.class);
        } finally {
            frame.release();
        }
    }
}
// 对应的编码器（使用LengthFieldPrepender）
public class OptimizedMessageEncoder2 extends MessageToByteEncoder<Message> {
    @Override
    protected void encode(ChannelHandlerContext ctx, Message msg, ByteBuf out) throws Exception {
        byte[] data = JSON.toJSONBytes(msg);
        out.writeInt(data.length);
        out.writeBytes(data);
    }
}

测试类

package com.example.netty;
import com.example.netty.codec.MessageDecoder;
import com.example.netty.codec.MessageEncoder;
import com.example.netty.protocol.Message;
import com.example.netty.protocol.MessageType;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.embedded.EmbeddedChannel;
import org.junit.jupiter.api.Test;
import static org.junit.jupiter.api.Assertions.*;
public class CodecTest {
    @Test
    public void testEncodeDecode() {
        // 创建EmbeddedChannel
        EmbeddedChannel channel = new EmbeddedChannel(new MessageDecoder());
        // 创建消息
        Message message = new Message(0xCAFEBABE, (byte) 1, MessageType.LOGIN_REQUEST.getCode(), 1001);
        message.put("username", "admin");
        message.put("password", "123456");
        // 测试编码
        ByteBuf buffer = Unpooled.buffer();
        MessageEncoder encoder = new MessageEncoder();
        try {
            // 手动执行编码（简化测试）
            encoder.encode(null, message, buffer);
            // 测试解码
            channel.writeInbound(buffer);
            // 读取解码结果
            Message decoded = channel.readInbound();
            assertNotNull(decoded);
            assertEquals(message.getMagicNumber(), decoded.getMagicNumber());
            assertEquals(message.getMessageType(), decoded.getMessageType());
            assertEquals(message.getSequenceId(), decoded.getSequenceId());
            assertEquals(message.get("username"), decoded.get("username"));
        } catch (Exception e) {
            fail("Codec test failed: " + e.getMessage());
        }
    }
    @Test
    public void testPartialData() {
        EmbeddedChannel channel = new EmbeddedChannel(new MessageDecoder());
        // 模拟分片数据
        byte[] fullData = createMessageData();
        int splitPoint = fullData.length / 2;
        // 第一次写入前半部分
        ByteBuf firstPart = Unpooled.wrappedBuffer(fullData, 0, splitPoint);
        channel.writeInbound(firstPart);
        // 此时不应该有完整的消息
        assertNull(channel.readInbound());
        // 写入剩余部分
        ByteBuf secondPart = Unpooled.wrappedBuffer(fullData, splitPoint, fullData.length - splitPoint);
        channel.writeInbound(secondPart);
        // 应该能读取到完整的消息
        assertNotNull(channel.readInbound());
    }
    private byte[] createMessageData() {
        ByteBuf buffer = Unpooled.buffer();
        MessageEncoder encoder = new MessageEncoder();
        Message message = new Message(0xCAFEBABE, (byte) 1, MessageType.CHAT_REQUEST.getCode(), 1002);
        message.put("content", "Test message");
        try {
            encoder.encode(null, message, buffer);
            byte[] data = new byte[buffer.readableBytes()];
            buffer.readBytes(data);
            return data;
        } catch (Exception e) {
            return new byte[0];
        }
    }
}

使用Protobuf的版本（高级）

// message.proto
syntax = "proto3";
package com.example.netty.protobuf;
message ProtoMessage {
    int32 magicNumber = 1;
    int32 version = 2;
    int32 messageType = 3;
    int32 sequenceId = 4;
    map<string, string> body = 5;
}

package com.example.netty.codec;
import com.example.netty.protobuf.ProtoMessage;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.handler.codec.MessageToByteEncoder;
import io.netty.handler.codec.ProtobufVarint32LengthFieldPrepender;
import io.netty.handler.codec.ProtobufVarint32FrameDecoder;
import io.netty.handler.codec.protobuf.ProtobufDecoder;
import io.netty.handler.codec.protobuf.ProtobufEncoder;
public class ProtobufCodec {
    // Protobuf编码器配置
    public static class ProtobufMessageEncoder extends MessageToByteEncoder<ProtoMessage.ProtoMessage> {
        @Override
        protected void encode(ChannelHandlerContext ctx, ProtoMessage.ProtoMessage msg, ByteBuf out) throws Exception {
            byte[] data = msg.toByteArray();
            out.writeBytes(data);
        }
    }
    // Protobuf解码器配置
    public static class ProtobufMessageDecoder extends ByteToMessageDecoder {
        @Override
        protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
            // 使用ProtobufVarinint32FrameDecoder处理
            // 这里需要配合ProtobufDecoder使用
        }
    }
    // 完整的pipeline配置
    public static void addProtobufCodec(ChannelPipeline pipeline) {
        pipeline.addLast(new ProtobufVarint32FrameDecoder());
        pipeline.addLast(new ProtobufDecoder(ProtoMessage.ProtoMessage.getDefaultInstance()));
        pipeline.addLast(new ProtobufVarint32LengthFieldPrepender());
        pipeline.addLast(new ProtobufEncoder());
    }
}

这个完整的Netty编解码案例包含了：