基于Netty的RPC简易实现

我就是我 2022-05-12 12:44 278阅读 0赞

> 代码地址如下：  
> [http://www.demodashi.com/demo/13448.html][http_www.demodashi.com_demo_13448.html]

> 可以给你提供思路
> 
> 也可以让你学到Netty相关的知识
> 
> 当然，这只是一种实现方式

### 需求 ###

看下图，其实这个项目就是为了做这样一件事。

有一个公共服务\*\*\*ServerA\*\*\*，它提供了一个名为\*\*\*getUserName\*\*\*的服务。

现在有多个类似**ServerB**的Web应用服务器。

当客户想通过**ServerB**要请求\*\**getUserName**服务时，由于**ServerB*服务中因为没有\*\*\*UserService\*\*\*的实现类，导致不能正常提供服务的问题。

![rpc原理][rpc]

### 预期结果 ###

可以看到，在Client项目中，UserService没有实现类，但是返回了正常的结果。

![运行结果][gZkryPtLvwhjgrWm2hR.png]

### 项目结构 ###

整个项目分为三个部分，**Server端、Client端以及一个公共jar。**

下图正是整个项目的目录结构图。

![目录结构][fYWmHn7giIfdSsfEDyE.jpg]

#### 公共部分 ####

公共部分的存在是因为我将服务器端和客户端写在了一个项目中，为了不让代码重复警告，所以提出来的一个公共模块。主要是几个实体类和一些序列化工具类。

*  MsgPackDecoder
    
     *  该类是一个MsgPack编码器，主要作用将Object对象序列化成byte数组。
 *  MsgPackEncoder
    
     *  该类是一个MsgPack解码器，主要作用将byte数组反序列化成Object对象。
 *  ObjectCodec
    
     *  该类是继承了MessageToMessageCodec<ByteBuf, Object>。是自定义序列化类主要有两个方法，encode和decode
        
            @Override protected void encode(ChannelHandlerContext ctx, Object msg, List<Object> out) { // 调用工具类的序列化方法 byte[] data = ObjectSerializerUtils.serilizer(msg); ByteBuf buf = Unpooled.buffer(); buf.writeBytes(data); out.add(buf); } @Override protected void decode(ChannelHandlerContext ctx, ByteBuf msg, List<Object> out) { // 调用工具类的反序列化方法 byte[] bytes = new byte[msg.readableBytes()]; msg.readBytes(bytes); Object deSerilizer = ObjectSerializerUtils.deSerilizer(bytes); out.add(deSerilizer); }
 *  ObjectSerializerUtils
    
     *  序列化工具类。序列化方法可以自己实现
 *  **MethodInvokeMeta**
    
     *  重点类。这个类是一个实体类。用于在网络中传输的类。主要有5个字段分别记录了一个接口的类对象，调用接口的方法名，方法的参数列表（包含参数类型，和参数列表），方法的返回值类型。
     *  在客户端中，这个类将调用方所要调用的方法封装。
     *  在服务端中，这个类主要用于服务器反射调用方法。
     *  当然，也可以用String来记录这些元信息。
 *  **NullWritable**
    
     *  这个类主要用于在网络中传输null。当返回值为null时，服务端会返回NullWritable对象。客户端接收到NullWritable时进行null处理。
 *  User
    
     *  实体对象。测试用例

#### 客户端 ####

上面的目录结构图也有提到，客户端中只有UserService接口，所以客户端中如果不做处理是不能正常运行的。

客户端中核心类有以下7个，其中与Netty相关的核心类与服务端一样有3个

*  ClientChannelHandlerAdapter
 *  CustomChannelInitializerClient
 *  NettyClient
 *  RpcProxyFactoryBean
 *  NettyBeanScanner
 *  PackageClassUtils
 *  WrapMethodUtils

**NettyClient**

这个类是Netty客户端的启动类，这个类中与Netty服务端进行通信

**CustomChannelInitializerClient**

这个类是用于初始化管道事件的类。主要添加了TCP粘包问题解决方案和自定义编解码器工具

**ClientChannelHandlerAdapter**

这个类是Netty客户端的处理类，主要通过这个类将调用信息写给Netty服务端。

**NettyBeanScanner**

这个类实现了BeanFactoryPostProcessor接口。BeanFactoryPostProcessor是Spring初始化Bean时对外暴露的扩展点。

> Spring IoC容器允许BeanFactoryPostProcessor在容器实例化任何bean之前读取bean的定义(配置元数据)，并可以修改它。同时可以定义多个BeanFactoryPostProcessor，通过设置’order’属性来确定各个BeanFactoryPostProcessor执行顺序。

在postProcessBeanFactory方法中，调用**PackageClassUtils.resolver**方法，将UserService.class注册到SpringBean工厂。

/** * 注册Bean到Spring的bean工厂 */ @Override public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException { this.beanFactory = (DefaultListableBeanFactory) beanFactory; // 加载远程服务的接口 List<String> resolverClass = PackageClassUtils.resolver(basePackage); for (String clazz : resolverClass) { String simpleName; if (clazz.lastIndexOf('.') != -1) { simpleName = clazz.substring(clazz.lastIndexOf('.') + 1); } else { simpleName = clazz; } BeanDefinitionBuilder gd = BeanDefinitionBuilder.genericBeanDefinition(RpcProxyFactoryBean.class); gd.addPropertyValue("interfaceClass", clazz); gd.addPropertyReference("nettyClient", clientName); this.beanFactory.registerBeanDefinition(simpleName, gd.getRawBeanDefinition()); } }

**RpcProxyFactoryBean**

重点类，这个类继承了AbstractFactoryBean实现了InvocationHandler。是一个代理类。

使用jdk动态代理的方式创建代理对象。

Proxy.newProxyInstance(interfaceClass.getClassLoader(), new Class[]{ interfaceClass}, this);

在invoke方法中，

调用**WrapMethodUtils**工具类中的方法，将代理对象方法封装成MethodInvokeMeta对象。

然后通过NettyClient传输给NettyServer端，进行RPC调用，并将结果返回。

至此，客户端的核心类介绍完毕。

#### 服务端 ####

服务端主要的核心类有三个

*  ServerChannelHandlerAdapter
 *  RequestDispatcher
 *  NettyServer

##### NettyServer #####

这个类主要有两个方法，一个是启动Netty服务的方法，一个是关闭服务器的方法。核心代码如下：

/** * 启动netty服务的方法 */ public void start() { // 服务器监听端口号 int port = serverConfig.getPort(); serverBootstrap.group(bossGroup, workerGroup) .channel(NioServerSocketChannel.class) // BACKLOG用于构造服务端套接字ServerSocket对象， // 标识当服务器请求处理线程全满时，用于临时存放已完成三次握手的请求的队列的最大长度。 // 如果未设置或所设置的值小于1，Java将使用默认值50 .option(ChannelOption.SO_BACKLOG, 100) .handler(new LoggingHandler(LogLevel.INFO)); try { // 设置事件处理 serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) { ChannelPipeline pipeline = ch.pipeline(); // 自定义长度解码器解决TCP黏包问题 // maxFrameLength 最大包字节大小，超出报异常 // lengthFieldOffset 长度字段的偏差 // lengthFieldLength 长度字段占的字节数 // lengthAdjustment 添加到长度字段的补偿值 // initialBytesToStrip 从解码帧中第一次去除的字节数 pipeline.addLast(new LengthFieldBasedFrameDecoder(65535 , 0, 2, 0, 2)); // LengthFieldPrepender编码器，它可以计算当前待发送消息的二进制字节长度，将该长度添加到ByteBuf的缓冲区头中 pipeline.addLast(new LengthFieldPrepender(2)); // 序列化工具 pipeline.addLast(new ObjectCodec()); pipeline.addLast(channelHandlerAdapter); } }); ChannelFuture f = serverBootstrap.bind(port).sync(); f.channel().closeFuture().sync(); } catch (InterruptedException e) { bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); } }

##### ServerChannelHandlerAdapter #####

这个类主要重写了ChannelRead方法，在ChannelRead方法中调用了RequestDispatcher类中的dispatcher方法来处理消息。

##### RequestDispatcher #####

这个类的作用为，将ChannelRead方法中读到的数据(也可以说命令)，通过反射来调用，执行对应方法，将执行后的结果写回通道，供客户端使用。

/** * 分发命令 * * @param channelHandlerContext channelHandlerContext * @param invokeMeta invokeMeta */ public void dispatcher(final ChannelHandlerContext channelHandlerContext, final MethodInvokeMeta invokeMeta) { ChannelFuture f = null; try { // 获取客户端准备调用的接口类 Class<?> interfaceClass = invokeMeta.getInterfaceClass(); // 获取准备调用的方法名称 String name = invokeMeta.getMethodName(); // 获取方法对应的参数列表 Object[] args = invokeMeta.getArgs(); // 获取参数类型 Class<?>[] parameterTypes = invokeMeta.getParameterTypes(); // 通过Spring获取对象 Object targetObject = app.getBean(interfaceClass); // 获取待调用方法 Method method = targetObject.getClass().getMethod(name, parameterTypes); // 执行方法 Object obj = method.invoke(targetObject, args); if (obj == null) { // 如果方法结果为空，将NULL结果写给客户端 f = channelHandlerContext.writeAndFlush(NullWritable.nullWritable()); } else { // 写给客户端结果 f = channelHandlerContext.writeAndFlush(obj); } // 释放通道，不是关闭连接 f.addListener(ChannelFutureListener.CLOSE); } catch (Exception e) { // 出现异常后的处理 f = channelHandlerContext.writeAndFlush(e.getMessage()); } finally { if (f != null) { f.addListener(ChannelFutureListener.CLOSE); } } }

### 使用方法 ###

1.  启动ServerApplication
2.  启动ClientApplication
3.  打开Chrome浏览器，输入：[http://localhost:8080/client/get/name][http_localhost_8080_client_get_name] 或者 [http://localhost:8080/client/get/info][http_localhost_8080_client_get_info] 即可看到结果。

如果想整合到现有项目中，请直接留言或者联系作者，此次并没有提供集成版本，但如果此篇文章已理解，那么自己可以手动的去集成到自己的项目中。基于Netty的RPC简易实现

> 代码地址如下：  
> [http://www.demodashi.com/demo/13448.html][http_www.demodashi.com_demo_13448.html]

> 注：本文著作权归作者，由demo大师发表，拒绝转载，转载需要作者授权

[http_www.demodashi.com_demo_13448.html]: http://www.demodashi.com/demo/13448.html
[rpc]: /images/20220512/d34459f80265487c9f5a7aeae932b64e.png
[gZkryPtLvwhjgrWm2hR.png]: /images/20220512/900efdbcdfd841dc87ff3029a5073304.png
[fYWmHn7giIfdSsfEDyE.jpg]: /images/20220512/9e21a26a6a324569b888be3d7b6a2b23.png
[http_localhost_8080_client_get_name]: http://localhost:8080/client/get/name
[http_localhost_8080_client_get_info]: http://localhost:8080/client/get/info