手撕RPC框架——引入Netty

本文最后更新于 2024年8月9日上午

上篇文章中我们使用了Socket编程实现了一个简单的RPC调用的demo，此篇文章，引入Netty框架来进一步提升框架性能。每篇文章对应的八股知识我会在收集整理相关资料后发出，敬请期待！😊

项目地址：【代码随想录知识星球】项目分享-手撕RPC框架

实现思路

Netty所起到的作用

客户端调用RpcClient.sendRequest方法 → NettyClientInitializer → Encoder编码 → 发送request
服务端RpcServer接收 → NettyServerInitializer → Decoder解码 → NettyRPCServerHandler → getResponse调用 → 返回response
客户端接收 → NettyServerInitializer → Decoder解码 → NettyRPCServerHandler处理结果并返回给上层

代码实现

在此部分我们将理论转化为代码，分别实现客户端和服务端。

项目目录结构

项目目录结构

定义RpcClient接口

public interface RpcClient {

    //定义底层通信的方法
    RpcResponse sendRequest(RpcRequest request);
}

配置netty对消息的处理机制

public class NettyClientInitializer extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        //使用长度字段解码器 消息格式 【长度】【消息体】，解决沾包问题
        pipeline.addLast(
                new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE,0,4,0,4));
        //计算当前待发送消息的长度，写入到前4个字节中
        pipeline.addLast(new LengthFieldPrepender(4));

        //使用Java序列化方式，netty的自带的解码编码支持传输这种结构
        pipeline.addLast(new ObjectEncoder());
        //使用了Netty中的ObjectDecoder，它用于将字节流解码为 Java 对象。
        //在ObjectDecoder的构造函数中传入了一个ClassResolver 对象，用于解析类名并加载相应的类。
        pipeline.addLast(new ObjectDecoder(new ClassResolver() {
            @Override
            public Class<?> resolve(String className) throws ClassNotFoundException {
                return Class.forName(className);
            }
        }));
        
        pipeline.addLast(new NettyClientHandler());
    }
}

Netty粘包问题的常见解决方案

粘包和拆包问题也叫做粘包和半包问题,它是指在数据传输时,接收方未能正常读取到一条完整数据的情况（只读取了部分数据,或多读取到了另一条数据的情况）就叫做粘包或拆包问题。

粘包问题

粘包问题是指在网络通信中，发送方连续发送的多个小数据包被接收方一次性接收的现象。这可能是因为底层传输层协议（如 TCP）会将多个小数据包合并成一个大的数据块进行传输，导致接收方在接收数据时一次性接收了多个数据包，造成粘连。

粘包问题实例

拆包/半包问题

拆包问题是指发送方发送的一个大数据包被接收方拆分成多个小数据包进行接收的现象。这可能是因为底层传输层协议（如 TCP）将一个大数据包拆分成多个小的数据块进行传输，导致接收方在接收数据时分别接收了多个小数据包，造成拆开。

拆包问题实例

xxxxxxxxxx public class TestServer { public static void main(String[] args) { UserService userService=new UserServiceImpl(); ServiceProvider serviceProvider=new ServiceProvider(); serviceProvider.provideServiceInterface(userService); RpcServer rpcServer=new SimpleRPCRPCServer(serviceProvider); rpcServer.start(9999); }}java

粘包问题通常发生在 TCP/IP 协议中，因为 TCP 是面向连接的传输协议，它是以“流”的形式传输数据的，而“流”数据是没有明确的开始和结尾边界的，所以就会出现粘包问题。

常见解决方案

固定大小方法：发送方和接收方固定发送数据大小，当字符长度不够时用空字符弥补，有了固定大小之后就知道每条消息的具体边界了，这样就没有粘包的问题了。
自定义数据协议（定义数据长度）：在 TCP 协议的基础上封装一层自定义数据协议，在自定义数据协议中，包含数据头（存储数据的大小）和数据的具体内容，这样服务端得到数据之后，通过解析数据头就可以知道数据的具体长度了，也就没有粘包的问题了。
特殊分割符：以特殊的字符结尾，比如以“\n”结尾，这样我们就知道数据的具体边界了，从而避免了粘包问题。

以上三种方案中，第一种固定大小的方法可能会造成网络流量的浪费，以及传输性能慢的问题；第二种解决方案实现难度大，且不利于维护，所以比较推荐的是第三种方案，使用特殊分隔符来区分消息的边界，从而避免粘包问题。

Netty解决方案

PS：在 Netty 中，解码器（Decoder）起着非常重要的作用。解码器主要负责将从网络中接收到的原始字节流数据转换为应用程序能够理解的 Java 对象或消息格式。使用解码器可以解决粘包和拆包问题、协议转换问题、消息编码（如文本转换为字节流）等问题。

使用定长解码器（FixedLengthFrameDecoder）：每个数据包都拥有固定的长度，接收端根据固定长度对数据进行切分，从而解决了粘包问题。
使用行分隔符解码器（LineBasedFrameDecoder）：以行为单位进行数据包的解码，从而解决粘包问题。
使用分隔符解码器（DelimiterBasedFrameDecoder）：使用特定的分隔符来标识消息边界，这样接收端可以根据分隔符正确切分消息。
使用长度字段解码器（LengthFieldBasedFrameDecoder）：在消息头部加入表示消息长度的字段，接收端根据长度字段来确定消息的边界，而从解决粘包问题。

NettyClientHandler类

指定对接收消息的处理方式

public class NettyClientHandler extends SimpleChannelInboundHandler<RpcResponse> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RpcResponse response) throws Exception {
        // 接收到response, 给channel设计别名，让sendRequest里读取response
        AttributeKey<RpcResponse> key = AttributeKey.valueOf("RPCResponse");
        //相当于将channel和特定信息进行绑定
        ctx.channel().attr(key).set(response);
        ctx.channel().close();
    }
    
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        //异常处理
        cause.printStackTrace();
        ctx.close();
    }
}

服务端重构

@AllArgsConstructor
public class NettyRPCRPCServer implements RpcServer {
    private ServiceProvider serviceProvider;
    @Override
    public void start(int port) {
        // netty 服务线程组boss负责建立连接， work负责具体的请求
        NioEventLoopGroup bossGroup = new NioEventLoopGroup();
        NioEventLoopGroup workGroup = new NioEventLoopGroup();
        System.out.println("netty服务端启动了");
        try {
            //启动netty服务器
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            //初始化
            serverBootstrap.group(bossGroup,workGroup).channel(NioServerSocketChannel.class)
                    //NettyClientInitializer这里 配置netty对消息的处理机制
                    .childHandler(new NettyServerInitializer(serviceProvider));
            //同步堵塞
            ChannelFuture channelFuture=serverBootstrap.bind(port).sync();
            //死循环监听
            channelFuture.channel().closeFuture().sync();
        }catch (InterruptedException e){
            e.printStackTrace();
        }finally {
            bossGroup.shutdownGracefully();
            workGroup.shutdownGracefully();
        }
    }

    @Override
    public void stop() {
    }
}

NettyRPCServerHandler类

NettyServerInitializer类和NettyClientInitializer类似，接收来自客户端的信息，并解析调用

@AllArgsConstructor
public class NettyRPCServerHandler extends SimpleChannelInboundHandler<RpcRequest> {
    private ServiceProvider serviceProvider;
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RpcRequest request) throws Exception {
        //接收request，读取并调用服务
        RpcResponse response = getResponse(request);
        ctx.writeAndFlush(response);
        ctx.close();
    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
    private RpcResponse getResponse(RpcRequest rpcRequest){
        //得到服务名
        String interfaceName=rpcRequest.getInterfaceName();
        //得到服务端相应服务实现类
        Object service = serviceProvider.getService(interfaceName);
        //反射调用方法
        Method method=null;
        try {
            method= service.getClass().getMethod(rpcRequest.getMethodName(), rpcRequest.getParamsType());
            Object invoke=method.invoke(service,rpcRequest.getParams());
            return RpcResponse.sussess(invoke);
        } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException e) {
            e.printStackTrace();
            System.out.println("方法执行错误");
            return RpcResponse.fail();
        }
    }
}

项目 > 手撕RPC框架

#Java #Netty #RPC #项目 #Socket

手撕RPC框架——引入Netty

https://love-enough.github.io/2024/07/07/手撕RPC框架——引入Netty/

作者

GuoZihan

发布于

2024年7月7日

更新于

2024年8月9日

许可协议

二分查找模板上一篇

手撕RPC框架——实现简单的RPC调用下一篇

手撕RPC框架——引入Netty

相关理论基础

Netty核心组件

Channel

EventLoop 与 EventLoopGroup

ServerBootstrap 与 Bootstrap

ChannelHandler 与 ChannelPipeline

ChannelFuture

与Socekt相比优势

实现思路

代码实现

定义RpcClient接口

配置netty对消息的处理机制

Netty粘包问题的常见解决方案

粘包问题

拆包/半包问题

常见解决方案

Netty解决方案

NettyClientHandler类

服务端重构

NettyRPCServerHandler类