自定义协议rpc netty

### 如何使用 Netty 实现自定义 RPC 协议 #### 设计自定义 RPC 协议的消息格式 为了实现高效的通信，首先需要设计一个清晰的协议消息格式。通常情况下，该格式可以包括以下几个部分：魔数、版本号、序列化算法标志位、请求 ID 和正文长度等字段[^1]。 ```java public class RpcProtocol<T> { private int magicNumber; // 魔数用于标识这是一个合法的RPC包 private byte version; // 版本号 private byte serializeType; // 序列化类型 private long requestId; // 请求ID private T body; // 正文数据 // Getter and Setter methods... } ``` #### 编解码器的设计与实现 在 Netty 中，编解码器负责将字节流转换成对象或将对象编码为字节流。可以通过继承 `ByteToMessageDecoder` 和 `MessageToByteEncoder` 来完成这一功能[^2]。 ##### 解码器实现 以下是基于上述消息格式的一个简单解码器示例： ```java import io.netty.buffer.ByteBuf; import io.netty.channel.ChannelHandlerContext; import io.netty.handler.codec.ByteToMessageDecoder; import java.util.List; public class RpcDecoder extends ByteToMessageDecoder { private Class<?> genericClass; public RpcDecoder(Class<?> genericClass) { this.genericClass = genericClass; } @Override protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception { if (in.readableBytes() < 4) { // 消息头至少有4个字节表示长度 return; } in.markReaderIndex(); int dataLength = in.readInt(); // 获取消息体长度 if (dataLength < 0) { ctx.close(); } if (in.readableBytes() < dataLength) { // 如果可读取的数据不足，则等待更多数据到达 in.resetReaderIndex(); return; } byte[] data = new byte[dataLength]; in.readBytes(data); Object obj = SerializationUtil.deserialize(data, genericClass); // 反序列化操作 out.add(obj); } } ``` ##### 编码器实现 同样地，也需要编写对应的编码器来处理发送到远程节点的数据： ```java import io.netty.buffer.Unpooled; import io.netty.channel.ChannelHandlerContext; import io.netty.handler.codec.MessageToByteEncoder; import java.nio.charset.StandardCharsets; public class RpcEncoder extends MessageToByteEncoder<RpcRequest> { @Override protected void encode(ChannelHandlerContext ctx, RpcRequest msg, io.netty.buffer.ByteBuf out) throws Exception { try { byte[] data = SerializationUtil.serialize(msg); // 将RpcRequest对象序列化 out.writeInt(data.length); // 写入消息体长度 out.writeBytes(data); // 写入实际数据 } catch (Exception e) { System.err.println("Encoding error: " + e.getMessage()); } } } ``` #### 启动服务端并绑定处理器 服务端通过 `ServerBootstrap` 类配置监听地址和端口，并注册管道中的各种处理器以响应客户端请求[^3]。 ```java package netty.dubborpc.server; 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; public class NettyServer { public static void startServer(String host, int port) { 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(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { ChannelPipeline pipeline = ch.pipeline(); // 添加编解码器和其他逻辑处理器 pipeline.addLast(new RpcDecoder(RpcRequest.class)); pipeline.addLast(new RpcEncoder()); pipeline.addLast(new RpcServerHandler()); // 处理业务逻辑 } }); ChannelFuture future = bootstrap.bind(host, port).sync(); System.out.println("Netty server started at " + host + ":" + port); future.channel().closeFuture().sync(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } finally { bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); } } } ``` #### 客户端连接和服务调用 客户端则需创建一个新的通道并与指定的服务端建立连接。之后即可利用此通道发起方法调用请求[^4]。 ```java package netty.dubborpc.client; import io.netty.bootstrap.Bootstrap; import io.netty.channel.*; import java.net.InetSocketAddress; public class NettyClient { private final String host; private final int port; public NettyClient(String host, int port) { this.host = host; this.port = port; } public Channel connect() throws InterruptedException { EventLoopGroup group = new NioEventLoopGroup(); Bootstrap bootstrap = new Bootstrap() .group(group) .channel(NioSocketChannel.class) .option(ChannelOption.TCP_NODELAY, true) .handler(new ChannelInitializer<NioSocketChannel>() { @Override protected void initChannel(NioSocketChannel ch) throws Exception { ChannelPipeline p = ch.pipeline(); // 注册编解码器及其他必要组件 p.addLast(new RpcDecoder(RpcResponse.class)); p.addLast(new RpcEncoder()); p.addLast(new RpcClientHandler()); } }); Channel channel = bootstrap.connect(new InetSocketAddress(host, port)).sync().channel(); System.out.println("Connected to the remote address [" + host + ":" + port + "]"); return channel; } } ``` ---