Netty实现RPC服务器之出、入站处理器

前言

本篇是用Netty实现一个RPC中间件的第三篇，未来还会持续更新，代码在最后一片更新完毕之后将会上传到github、gitee代码托管平台，供大家拉取学习，并进行讨论分享，让该RPC框架持续精进。

在Netty中，数据的处理是通过处理器链（pipeline）完成的，这些处理器可以被分类为出站处理器（outbound handlers）和入站处理器（inbound handlers）。出站处理器负责处理从应用程序向网络发送的数据，而入站处理器则处理从网络接收到的应用程序数据。

本篇主要接扫通过构建自定义的处理器，让我们的消息在发送、接收阶段进行一定的逻辑处理，包括心跳处理，编解码，业务处理器等等。

愿我们在技术的浩瀚中游刃有余。

软件中目录

rpc_core模块

rpc_client模块

rpc_server模块

出/入站处理器

概述

netty使用了一种事件驱动的应用范式，因此数据处理的管道是一个经过处理器的事件链。事件和处理器可以关联到 xxInbound入站 与 xxOutbound出站 数据流。

处理器链

入站和出栈事件都会经过预设的处理器链（多个处理器）；

• 即入站事件经过 入站处理器；出站事件经过出站处理器；多个处理器形成一个链或管道；

处理器举例：

网络传输全是字节形式，而业务逻辑处理是对象形式，所以需要编码器把对象转字节，需要解码器把字节转对象；

• ByteToMessageDecoder 字节转消息（对象）解码器；
• MessageToByteEncoder 消息（对象）转字节编码器；
• 业务逻辑处理器（如加工，统计，入库，消息转发等）；

以客户端服务器模式介绍入站与出站处理器的事件处理过程

【图解】

客户端的处理器有：

• 解码处理器；
• 编码处理器；
• 客户端业务处理器；

服务端的处理器有：

• 解码处理器；
• 编码处理器；
• 服务器业务处理器；

补充：多个处理器封装到通道管道 ChannelPipeline；

处理器链调用机制代码实现

1）需求描述：

自定义编码器和解码器实现客户端与服务器间的数据传输； 2）通道管道``ChannelPipeline` 可以封装多个处理器；其处理器执行顺序特别重要（前后关系特别重要，如入栈解码处理器要第1个执行，又如出站编码器要最后一个执行），否则客户端与服务器将无法通信（因为事件或数据要经过所有的处理器）；类似于如下：

java

代码解读

复制代码

for (event event : events) { handler1(event); handler2(event); handler3(event); }

3）入站与出站处理器执行顺序：

3.1）服务器初始化器，添加处理器到管道； 

入站事件

Netty中的入站事件是指那些从网络流向应用程序的数据或状态变化。这些事件通常是由网络I/O操作触发的，然后由Netty框架传递给用户定义的处理器进行处理。以下是Netty中的一些典型入站事件：

1. ChannelRegistered：
   • 当Channel被注册到EventLoop之后触发。
   • 这个时候Channel还不能进行I/O操作，因为还没有完成绑定。
2. ChannelUnregistered：
   • 当Channel从EventLoop注销时触发。
   • 这通常发生在Channel关闭之前。
3. ChannelActive：
   • 当Channel准备好进行I/O操作时触发。
   • 对于客户端来说，这意味着已经成功连接到了服务器；对于服务器来说，这意味着一个新的连接已经被接受。
4. ChannelInactive：
   • 当Channel不再能够进行I/O操作时触发。
   • 这意味着Channel已经关闭或者连接已经断开。
5. ChannelRead：
   • 当从Channel中读取到数据时触发。
   • 这个事件将读取的数据传递给下一个Inbound Handler。
6. ChannelReadComplete：
   • 在所有ChannelRead事件都已被触发且所有数据都被读取完毕后触发。
   • 这是一个提示信号，表明当前Buffer中的所有数据都已经读取完成。
7. UserEventTriggered：
   • 用户可以自定义事件，这些事件也会被传递给Inbound Handler。
   • 常见的用途包括心跳检测、定时任务等。
8. ExceptionCaught：
   • 当Channel或Handler抛出一个未处理的异常时触发。
   • 这个事件允许你捕获异常并进行适当的错误处理。

在Netty中，这些事件会被传递给ChannelPipeline中的Inbound Handlers，这些Handlers按照它们在Pipeline中的顺序被调用。每个Handler都有机会处理事件，并可以选择继续传递给下一个Handler或停止传递。这允许开发者构建复杂的事件处理逻辑，同时保持代码的模块化和易于维护。

示例：处理ChannelRead事件

假设你需要实现一个入站处理器，该处理器用于解析接收到的消息：

整理了一份好像面试笔记包括了：Java面试、Spring、JVM、MyBatis、Redis、MySQL、并发编程、微服务、Linux、Springboot、SpringCloud、MQ、Kafka 面试专题

需要全套面试笔记【点击此处即可】免费获取

java

代码解读

复制代码

public class MessageDecoderHandler extends ChannelInboundHandlerAdapter { @Override public void channelRead(ChannelHandlerContext ctx, Object msg) { // 检查msg是否是我们需要处理的对象类型 if (msg instanceof ByteBuf) { ByteBuf in = (ByteBuf) msg; try { // 解析ByteBuf中的数据 String receivedMessage = in.toString(CharsetUtil.UTF_8); System.out.println("Received: " + receivedMessage); // 处理数据，例如转发给其他组件 } finally { // 释放ByteBuf资源 in.release(); } } else { // 如果不是我们处理的对象类型，则直接传递给下一个处理器 ctx.fireChannelRead(msg); } } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { // 打印异常信息 cause.printStackTrace(); // 关闭连接 ctx.close(); } }

在这个例子中，MessageDecoderHandler会在接收到数据时解析ByteBuf中的内容，并打印出来。如果解析完成后，会释放ByteBuf资源以避免内存泄漏。

将入站处理器添加到Pipeline

要将入站处理器添加到ChannelPipeline中，可以使用如下方法：

java

代码解读

复制代码

channel.pipeline().addLast(new MessageDecoderHandler());

通过这种方式，你可以根据需要添加多个入站处理器，并且这些处理器会按照它们在Pipeline中的顺序被调用。这种设计使得网络编程更加模块化和灵活。

出站事件

在Netty中，出站处理器（Outbound Handlers）主要负责处理应用程序发起的网络请求或数据发送。出站处理器通常是按照相反的顺序执行的，即它们按照在ChannelPipeline中的逆序被调用。以下是一些常见的出站事件及其处理方式：

1. ChannelRegistered：
   • 当Channel被注册到EventLoop之后触发。
   • 这个事件对出站处理器来说可能不是特别重要，但在某些场景下可能需要做一些初始化工作。
2. ChannelActive：
   • 当Channel准备好进行I/O操作时触发。
   • 对于出站处理器而言，这意味着可以开始发送数据了。
3. ChannelWritabilityChanged：
   • 当Channel的写入能力发生变化时触发。
   • 例如，当Channel由于背压（backpressure）而变得不可写入时，这个事件会被触发。
   • 出站处理器可以在这个事件中决定是否继续写入数据。
4. ChannelWrite：
   • 当应用程序尝试通过ChannelHandlerContext.write()方法写入数据时触发。
   • 这个事件会将要发送的数据传递给出站处理器。
5. ChannelFlushed：
   • 当ChannelHandlerContext.flush()被调用时触发。
   • 这表示所有的待写入的数据应该被尽快写出到网络中。

处理出站事件

出站处理器通常会实现ChannelOutboundHandler接口或其子接口。以下是如何处理这些事件的一些示例：

示例：处理ChannelWrite事件

假设你需要实现一个出站处理器，该处理器在数据被发送前对其进行加密：

java

代码解读

复制代码

public class EncryptionOutboundHandler extends ChannelOutboundHandlerAdapter { @Override public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception { // 检查msg是否是我们需要加密的对象类型 if (msg instanceof String) { String message = (String) msg; byte[] encryptedMessage = encrypt(message); // 自定义的加密逻辑 ByteBuf encoded = ctx.alloc().buffer(); encoded.writeBytes(encryptedMessage); ctx.write(encoded, promise); } else { // 如果不是我们处理的对象类型，则直接传递给下一个处理器 ctx.write(msg, promise); } } private byte[] encrypt(String message) { // 加密逻辑 return message.getBytes(); // 示例中简单地转换为字节数组 } @Override public void flush(ChannelHandlerContext ctx) throws Exception { ctx.flush(); } }

在这个例子中，EncryptionOutboundHandler会在数据被发送之前进行加密处理，并使用ctx.write()方法将加密后的数据传递给下一个出站处理器。如果没有需要加密的对象，则直接传递给下一个处理器。

将出站处理器添加到Pipeline

要将出站处理器添加到ChannelPipeline中，可以使用如下方法：

java

代码解读

复制代码

channel.pipeline().addLast(new EncryptionOutboundHandler());

RPC出/入站处理器

服务器端的RpcRequestMessageHandler处理器

java

代码解读

复制代码

/** * RPC请求处理器 * @author XiaoSheng * @date 2024/8/21 上午11:26 */ @Slf4j @ChannelHandler.Sharable public class RpcRequestMessageHandler extends SimpleChannelInboundHandler { @Override protected void channelRead0(ChannelHandlerContext ctx, RpcRequestMessage msg) throws Exception { RpcResponseMessage responseMessage = new RpcResponseMessage(); // 设置请求的序号 responseMessage.setSequenceId(msg.getSequenceId()); Object result; try { //通过名称从工厂获取本地注解了@RpcServer的实例, 通过服务类名称获取对应的类实例 Object service = ServiceFactory.serviceFactory.get(msg.getServiceName()); // 获取方法，参数， Method method = service.getClass().getMethod(msg.getMethodName(), msg.getParameterTypes()); // 调用 result = method.invoke(service, msg.getParameterValue()); System.out.println("result: " + result); // 此处对该对象进行toString编码,防止序列化等原因导致复杂对象消息无法传递至客户端 responseMessage.setReturnValue(result.toString()); } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException e) { responseMessage.setExceptionValue(new Exception("远程调用出错: "+ e.getMessage())); } finally { ctx.writeAndFlush(responseMessage); /** * ReferenceCountUtil.release 方法是用来减少引用计数的。 * 当一个引用计数对象不再被需要时，你应该调用这个方法来减少它的引用计数。 * 如果引用计数减为零，Netty 将会自动释放该对象的内存。 */ ReferenceCountUtil.release(msg); } } /** * 读空闲 * @param ctx * @param evt * @throws Exception */ @Override public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { if (evt instanceof IdleStateEvent) { IdleState state = ((IdleStateEvent) evt).state(); if (state == IdleState.READER_IDLE) { log.info("长时间未收到心跳包，断开连接..."); ctx.close(); } } else { super.userEventTriggered(ctx, evt); } } }

客户端的RpcResponseMessageHandler处理器

java

代码解读

复制代码

/** * @author XiaoSheng * @date 2024/8/22 上午11:48 */ @Slf4j @ChannelHandler.Sharable public class RpcResponseMessageHandler extends SimpleChannelInboundHandler { @Override protected void channelRead0(ChannelHandlerContext ctx, RpcResponseMessage msg) throws Exception { try { log.debug("{}", msg); // 每次使用完都要移除 Promise