基于Netty实现简单的RPC框架

本是古典 何须时尚 2022-08-29 10:28 379阅读 0赞

Netty实现RPC框架

关于 Netty 的线程模型，详见博客，传送地址：Netty线程模型

文章目录

  • Netty实现RPC框架
    • 一、RPC简介
    • 二、代码实现
        1. 需求介绍
        1. 实现步骤
        1. 公共代码
        1. 服务端代码
        1. 客户端代码

一、RPC简介

RPC全称为remote procedure call，即远程过程调用。借助RPC可以做到像本地调用一样调用远程服务，是一种进程间的通信方式。

比如两台服务器A和B，A服务器上部署一个应用，B服务器上部署一个应用，A服务器上的应用想调用B服务器上的应用提供的方法。由于两个应用不在一个内存空间，不能直接调用，所以需要通过网络来表达调用的语义和传达调用的数据。

注意：RPC并不是一个具体的技术，而是指整个网络远程调用过程。

image-20210713170847318

二、代码实现

1. 需求介绍

要求用Netty实现一个简单的RPC框架，消费者和提供者约定接口和协议，消费者远程调用提供者的服务。

2. 实现步骤

  1. 创建一个接口，定义抽象方法。用于消费者和提供者之间的约定。
  2. 创建一个提供者，该类需要监听消费者的请求，并按照约定返回数据。
  3. 创建一个消费者，该类需要透明的调用自己不存在的提供者的方法，内部需要使用Netty进行数据通信。
  4. 提供者与消费者数据传输使用 json 字符串数据格式。
  5. 使用 Netty 集成 SpringBoot 环境实现。

image-20210713184235953

注意：服务端注解要使用在实现类上，用于标记服务是否对外暴露。

3. 公共代码

  1. 公共接口，消费者与提供者的约定

    public interface IUserService { 

    1. /** * 根据ID查询用户 */
    2. User getById(int id);

    }

  2. 封装的请求对象

    @Data
    public class RpcRequest { 

    1. /** * 请求对象的ID */
    2. private String requestId;
    3. /** * 要调用方法所在类名 */
    4. private String className;
    5. /** * 要调用的方法名 */
    6. private String methodName;
    7. /** * 要调用方法的参数类型 */
    8. private Class<?>[] parameterTypes;
    9. /** * 要调用方法的入参 */
    10. private Object[] parameters;

    }

  3. 封装的响应对象

    @Data
    public class RpcResponse { 

    1. /** * 响应ID */
    2. private String requestId;
    4. /** * 错误信息 */
    5. private String error;
    7. /** * 返回的结果 */
    8. private Object result;

    }

  4. pojo类

    @Data
    public class User { 

    1. private int id;
    2. private String name;

    }

4. 服务端代码

  1. 定义注解，用于暴露服务接口，暴露服务端的getUserById方法

    @Target(ElementType.TYPE) //用于类上
    @Retention(RetentionPolicy.RUNTIME) //在运行时可以获取到
    public @interface RpcService {
    }

  2. 服务端实现公共接口，客户端要调用的方法

    @RpcService //对外暴露，加自定义注解
    @Service
    public class UserServiceImpl implements IUserService { 

    1. Map<Object, User> userMap = new HashMap();
    3. @Override
    4. public User getById(int id) { 
    5.     if (userMap.size() == 0) { 
    6.         User user1 = new User();
    7.         user1.setId(1);
    8.         user1.setName("张三");
    9.         User user2 = new User();
    10.         user2.setId(2);
    11.         user2.setName("李四");
    12.         userMap.put(user1.getId(), user1);
    13.         userMap.put(user2.getId(), user2);
    14.     }
    15.     return userMap.get(id);
    16. }

    }

  3. 服务端代码

    /* Netty的服务端 启动服务端监听端口 实现Spring容器生命周期接口DisposableBean，在Spring容器关闭的时候本类中的资源也要关闭 */
    @Component
    public class NettyRpcServer implements DisposableBean { 

    1. /** * 注入自定义的处理器 */
    2. @Autowired
    3. NettyServerHandler nettyServerHandler;
    5. // 定义boss和worker线程
    6. EventLoopGroup bossGroup = null;
    7. EventLoopGroup workerGroup = null;
    9. /** * 启动服务端，参数表示ip和端口号 */
    10. public void start(String host, int port) { 
    11.     try { 
    12.         //1.创建bossGroup和workerGroup
    13.         bossGroup = new NioEventLoopGroup(1);
    14.         workerGroup = new NioEventLoopGroup();
    16.         //2.设置启动助手
    17.         ServerBootstrap bootstrap = new ServerBootstrap();
    19.         //3.设置启动参数
    20.         bootstrap.group(bossGroup, workerGroup)
    21.             .channel(NioServerSocketChannel.class)
    22.             .childHandler(new ChannelInitializer<SocketChannel>() { 
    23.                 @Override
    24.                 protected void initChannel(SocketChannel ch) throws Exception { 
    25.                     //添加String的编解码器，使得客户端和服务器可以直接使用String类型进行交流
    26.                     ch.pipeline().addLast(new StringDecoder());
    27.                     ch.pipeline().addLast(new StringEncoder());
    29.                     //添加自定义处理器
    30.                     ch.pipeline().addLast(nettyServerHandler);
    31.                 }
    32.             });
    34.         //绑定ip和端口号
    35.         ChannelFuture channelFuture = bootstrap.bind(host, port).sync();
    37.         System.out.println("======Netty服务端启动成功======");
    39.         //监听通道的关闭状态（此处并没有真正的关闭），阻塞直到close事件发生
    40.         channelFuture.channel().closeFuture().sync();
    42.     } catch (InterruptedException e) { 
    43.         e.printStackTrace();
    44.         //关闭资源
    45.         if (bossGroup != null) { 
    46.             bossGroup.shutdownGracefully();
    47.         }
    48.         if (workerGroup != null) { 
    49.             workerGroup.shutdownGracefully();
    50.         }
    51.     }
    52. }
    54. /** * 实现DisposableBean接口要重写的方法，Spring关闭的时候执行 */
    55. @Override
    56. public void destroy() throws Exception { 
    57.     //关闭资源
    58.     if (bossGroup != null) { 
    59.         bossGroup.shutdownGracefully();
    60.     }
    61.     if (workerGroup != null) { 
    62.         workerGroup.shutdownGracefully();
    63.     }
    64. }

    }

  4. 服务端自定义处理器

    /* 服务端自定义处理器 1.将标有@RpcService注解的bean进行缓存，方便后续查找调用 2.接收客户端的请求 3.根据传递过来的beanName从缓存中查找bean 4.通过反射调用bean的方法 5.给客户端响应 /
    @Component
    @ChannelHandler.Sharable //设置通道共享，NettyServerHandler默认是单例的，不能由多个客户端所共享
    public class NettyServerHandler extends SimpleChannelInboundHandler implements ApplicationContextAware { 

    1. /** * 定义缓存，将将标有@RpcService注解的bean缓存至此map */
    2. static Map<String, Object> SERVICE_INSTANCE_MAP = new HashMap<>();
    4. /** * 1.将标有@RpcService的注解的bean进行缓存，实现ApplicationContextAware接口后重写的方法 * @param applicationContext ioc容器 */
    5. @Override
    6. public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { 
    7.     //1.1 通过注解获取bean的集合（可能有多个类都使用了这个注解）
    8.     Map<String, Object> serviceMap = applicationContext.getBeansWithAnnotation(RpcService.class);
    10.     //1.2 循环遍历
    11.     Set<Map.Entry<String, Object>> entries = serviceMap.entrySet();
    12.     for (Map.Entry<String, Object> entry : entries) { 
    14.         //serviceBean就是自定义的UserServiceImpl的对象
    15.         Object serviceBean = entry.getValue();
    16.         if (serviceBean.getClass().getInterfaces().length == 0) { 
    17.             throw new RuntimeException("对外暴露的服务必须实现接口");
    18.         }
    20.         //默认处理第一个接口作为缓存bean的名字（一个类可以实现多个接口）
    21.         String serviceName = serviceBean.getClass().getInterfaces()[0].getName();
    22.         SERVICE_INSTANCE_MAP.put(serviceName, serviceBean);
    24.         System.out.println(SERVICE_INSTANCE_MAP);
    25.         //{包名.IUserService=包名.UserServiceImpl@44c73c26}
    26.     }
    27. }
    29. /** * 2.接收客户端的请求，通道读取就绪事件，读取客户端的消息 * @param ctx 每一个Handler实例与Pipeline之间的桥梁就是ChannelHandlerContext实例 * @param msg 客户端发送过来的消息，是一个JSON格式的RpcRequest的对象 */
    30. @Override
    31. protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception { 
    33.     // 接收客户端的请求
    34.     RpcRequest rpcRequest = JSON.parseObject(msg, RpcRequest.class);
    36.     //创建响应对象，请求id是请求中传递过来的
    37.     RpcResponse rpcResponse = new RpcResponse();
    38.     rpcResponse.setRequestId(rpcRequest.getRequestId());
    40.     //调用自定义的业务处理方法（上述的3、4步），并将处理结果保存到响应中
    41.     try { 
    42.         rpcResponse.setResult(handler(rpcRequest));
    43.     } catch (Exception e) { 
    44.         e.printStackTrace();
    45.         //业务处理过程中出现的异常也要返回给客户端
    46.         rpcResponse.setError(e.getMessage());
    47.     }
    49.     //5. 将响应对象给客户端响应
    50.     ctx.writeAndFlush(JSON.toJSONString(rpcResponse));
    51. }
    53. /** * 定义业务处理方法（上述的3、4步） */
    54. private Object handler(RpcRequest rpcRequest) throws InvocationTargetException { 
    55.     //3.根据传递过来的beanName从缓存中查找，className是请求对象中的属性
    56.     Object serviceBean = SERVICE_INSTANCE_MAP.get(rpcRequest.getClassName());
    57.     if (serviceBean == null) { 
    58.         throw new RuntimeException("服务端没有找到服务");
    59.     }
    61.     // 4.通过反射调用bean的方法
    62.     // 调用cglid提供的FastClass中的方法，创建指定类的代理对象
    63.     FastClass proxyClass = FastClass.create(serviceBean.getClass());
    65.     //参数分别表示客户端调用的服务端的方法和参数类型，通过返回的method完成最终的方法调用
    66.     FastMethod method = proxyClass.getMethod(rpcRequest.getMethodName(), rpcRequest.getParameterTypes());
    67.     return method.invoke(serviceBean, rpcRequest.getParameters()); //反射
    68. }

    }

通过上述代码可知，必须加了自定义注解 @RpcService 的 bean 才会被缓存（服务才可以被调用），即自定义注解用来暴露服务，且服务端必须实现接口，约定协议，否则抛出异常。

  1. 服务端启动类

    /* 项目启动的时候需要将服务端启动并监听端口 启动类实现CommandLineRunner接口 CommandLineRunner接口是在容器启动成功后的最后一步回调（类似开机自启动） */
    @SpringBootApplication
    public class ServerBootstrapApplication implements CommandLineRunner { 

    1. /** * 注入Netty服务端 */
    2. @Autowired
    3. NettyRpcServer rpcServer;
    5. public static void main(String[] args) { 
    6.     SpringApplication.run(ServerBootstrapApplication.class, args);
    7. }
    9. /** * 实现CommandLineRunner接口后要重写的方法，创建线程调用服务端的方法启动服务端 */
    10. @Override
    11. public void run(String... args) throws Exception { 
    12.     new Thread(new Runnable() { 
    13.         @Override
    14.         public void run() { 
    15.             rpcServer.start("127.0.0.1", 8899);
    16.         }
    17.     }).start();
    18. }

    }

流程总结：

image-20210713183843918

5. 客户端代码

  1. 客户端业务处理类，向服务端发送数据并接收返回结果

    @Component
    public class NettyRpcClientHandler extends SimpleChannelInboundHandler implements Callable { 

    1. ChannelHandlerContext context; //全局的ChannelHandlerContext对象
    3. private String reqMsg; //发送的消息
    5. private String respMsg; //接收的消息
    7. //给发送的消息赋值
    8. public void setReqMsg(String reqMsg) { 
    9.     this.reqMsg = reqMsg;
    10. }
    12. /** * 通道读取就绪事件，读取服务端消息 * @param msg 接收到的服务端数据 */
    13. @Override
    14. protected synchronized void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception { 
    15.     respMsg = msg;
    16.     //唤醒call方法的等待线程
    17.     notify();
    18. }
    20. /*** * 通道连接就绪事件，给全局的ChannelHandlerContext对象赋值 * 不能连接建立之后就给服务端发送消息，因为不知道要发送什么消息，客户端调用controller方法才可发送消息，所以实现Callable接口，调用call方法时才发送消息 */
    21. @Override
    22. public void channelActive(ChannelHandlerContext ctx) throws Exception { 
    23.     context = ctx;
    24. }
    26.  /*** * 使用线程的等待与唤醒完成消息的同步处理，因为netty仅支持异步，不能发送完消息然后等待服务端的消息返回。 */
    28. /** * 实现Callable接口重写的方法 * 通过全局的ChannelHandlerContext对象给服务端发送消息，调用call方法才会发送 */
    29. @Override
    30. public synchronized Object call() throws Exception { 
    31.     context.writeAndFlush(reqMsg);
    32.     //客户端向服务端发送消息之后，要等待服务端的响应，调用本类的channelRead0方法读取服务端的数据
    33.     //将线程处于等待状态
    34.     wait();
    35.     return respMsg; //被唤醒之后，返回从服务端接收到的消息
    36. }

    }

  2. 客户端代码

    /* Netty客户端 1.连接服务端，工程一启动就去连接服务端，实现InitializingBean接口 2.提供发送消息的方法 3.工程停止后关闭资源，实现DisposableBean接口 /
    @Component
    public class NettyRpcClient implements InitializingBean, DisposableBean { 

    1. EventLoopGroup group = null;
    2. Channel channel = null;
    4. //注入自定义的客户端处理器
    5. @Autowired
    6. NettyRpcClientHandler nettyRpcClientHandler;
    8. //创建线程池，用于执行自定义处理器中重写的call方法，向服务端发送消息
    9. ExecutorService service = Executors.newCachedThreadPool();
    11. /** * 实现InitializingBean接口要重写的方法 * 1.连接服务端 */
    12. @Override
    13. public void afterPropertiesSet() throws Exception { 
    14.     try { 
    15.         //1.1 创建线程组
    16.         group = new NioEventLoopGroup();
    17.         //1.2 创建客户端启动助手
    18.         Bootstrap bootstrap = new Bootstrap();
    19.         //1.3 设置参数
    20.         bootstrap.group(group)
    21.                 .channel(NioSocketChannel.class)
    22.                 .handler(new ChannelInitializer<SocketChannel>() { 
    23.                     @Override
    24.                     protected void initChannel(SocketChannel ch) throws Exception { 
    25.                         //添加编解码器
    26.                         ch.pipeline().addLast(new StringDecoder());
    27.                         ch.pipeline().addLast(new StringEncoder());
    28.                         //添加自定处理类
    29.                         ch.pipeline().addLast(nettyRpcClientHandler);
    30.                     }
    31.                 });
    32.         //1.4 连接服务端
    33.         channel = bootstrap.connect("127.0.0.1", 8899).sync().channel();
    34.     } catch (Exception e) { 
    35.         e.printStackTrace();
    36.         if (channel != null) { 
    37.             channel.close();
    38.         }
    39.         if (group != null) { 
    40.             group.shutdownGracefully();
    41.         }
    42.     }
    43. }
    45. /** * 3.工程停止后关闭资源 * 实现DisposableBean接口要重写的方法 */
    46. @Override
    47. public void destroy() throws Exception { 
    48.     if (channel != null) { 
    49.         channel.close();
    50.     }
    51.     if (group != null) { 
    52.         group.shutdownGracefully();
    53.     }
    54. }
    56. /** * 2. 消息发送的方法，调用自定义处理器中的方法定义发送的消息 * @param msg 要发送的消息 * @return 服务端返回的消息 */
    57. public Object send(String msg) throws ExecutionException, InterruptedException { 
    59.     //调用自定义处理器的方法给要发送的消息赋值
    60.     nettyRpcClientHandler.setReqMsg(msg);
    62.     //执行自定义处理器中的call方法，向服务端发送消息
    63.     Future submit = service.submit(nettyRpcClientHandler);
    65.     //得到call方法的返回值，也就是服务端返回的消息
    66.     return submit.get();
    67. }

    }

  3. 公共接口IUserService的代理

    调用 IUserService 接口的 getById() 方法时，自动的封装RpcRequest，不然每调用一次方法就手动封装一个RpcRequest对象，不合理。

    而且不论其他的controller访问公共接口时，不用再去创建公共接口，直接访问缓存中的代理对象即可。

    /* IUserService的代理 1. 发送消息之前将消息封装成RpcRequest对象，并发送消息 2. 解析服务器返回的数据 */
    @Component
    public class RpcClientProxy { 

    1. @Autowired
    2. NettyRpcClient rpcClient;
    4. //缓存创建好的代理对象，方便复用
    5. Map<Class, Object> SERVICE_PROXY = new HashMap<>();
    7. /** * 获取代理对象，参数表示获取谁的代理对象 */
    8. public Object getProxy(Class serviceClass) { 
    10.     //从缓存中查找
    11.     Object proxy = SERVICE_PROXY.get(serviceClass);
    13.     if (proxy == null) { 
    14.         //创建代理对象
    15.         //第二参数表示IUserService
    16.         proxy = Proxy.newProxyInstance(this.getClass().getClassLoader(),
    17.                 new Class[]{ serviceClass}, new InvocationHandler() { 
    19.                     //invoke方法不会被立即调用，只有访问接口的方法时才会调用
    20.                     @Override
    21.                     public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { 
    22.                         //1.封装请求对象
    23.                         RpcRequest rpcRequest = new RpcRequest();
    24.                         rpcRequest.setRequestId(UUID.randomUUID().toString());
    26.                         //method是接口的方法封装后的对象，getclass得到的就是接口，而服务端的缓存bean的名字就是接口的名字
    27.                         rpcRequest.setClassName(method.getDeclaringClass().getName());
    28.                         rpcRequest.setMethodName(method.getName());
    29.                         rpcRequest.setParameterTypes(method.getParameterTypes());
    30.                         rpcRequest.setParameters(args);
    32.                         try { 
    33.                             //2.发送消息，返回值表示服务端返回的数据
    34.                             Object msg = rpcClient.send(JSON.toJSONString(rpcRequest));
    36.                             //3.将服务端的消息转化
    37.                             RpcResponse rpcResponse = JSON.parseObject(msg.toString(), RpcResponse.class);
    38.                             if (rpcResponse.getError() != null) { 
    39.                                 throw new RuntimeException(rpcResponse.getError());
    40.                             }
    41.                             if (rpcResponse.getResult() != null) { 
    42.                                 return JSON.parseObject(rpcResponse.getResult().toString(),
    43.                                         method.getReturnType());
    44.                             }
    45.                             return null;
    46.                         } catch (Exception e) { 
    47.                             e.printStackTrace();
    48.                             throw e;
    49.                         }
    50.                     }
    51.                 });
    53.         //将代理对象放入缓存
    54.         SERVICE_PROXY.put(serviceClass, proxy);
    55.         return proxy;
    56.     } else { 
    57.         return proxy;
    58.     }
    59. }

    }

  4. 引用代理类

    /* 引用代理类，使用在controller的IUserService上 */
    @Target(ElementType.FIELD) //作用于字段
    @Retention(RetentionPolicy.RUNTIME) //在运行时可以获取得到
    public @interface RpcReference {
    }

  5. 用户控制类

    /* 用户控制类，远程调用服务端的getUserById方法 */
    @RestController
    @RequestMapping(“/user”)
    public class UserController { 

    1. //使用了注解之后，目前还不能使用代理对象，必须经过后续的处理才能将代理对象注入进来
    2. @RpcReference
    3. IUserService userService;
    5. @RequestMapping("/getUserById")
    6. public User getUserById(int id) { 
    7.     //调用服务端方法
    8.     return userService.getById(id);
    9. }

    }

  6. bean的后置增强，将代理对象注入到使用了自定义的 @RpcReference 注解的字段上

    类似于 @Autowired 注解，使用这个注解之后将某个对象注入到这个字段上

    /* bean的后置增强 */
    @Component
    public class MyBeanPostProcessor implements BeanPostProcessor { 

    1. @Autowired
    2. RpcClientProxy rpcClientProxy;
    4. /** * 自定义注解的注入 */
    5. @Override
    6. public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException { 
    7.     //1.得到bean（u）中的所有字段
    8.     Field[] declaredFields = bean.getClass().getDeclaredFields();
    10.     //遍历字段
    11.     for (Field field : declaredFields) { 
    13.         //2.查找字段中是否包含RpcReference这个注解
    14.         RpcReference annotation = field.getAnnotation(RpcReference.class);
    16.         if (annotation != null) { 
    18.             //3.获取代理对象
    19.             Object proxy = rpcClientProxy.getProxy(field.getType());
    21.             try { 
    22.                 //4.属性注入，field表示加了RpcReference注解的IUserService对象
    23.                 field.setAccessible(true);
    24.                 field.set(bean, proxy); //userService由代理对象创建
    25.             } catch (IllegalAccessException e) { 
    26.                 e.printStackTrace();
    27.             }
    28.         }
    29.     }
    30.     return bean;
    31. }

    }

流程总结：

image-20210713211703690

运行结果：

将服务端先启动，然后启动客户端，进行访问：

image-20210713212210086

发表评论

表情：
评论列表 （有 0 条评论，379人围观）

还没有评论，来说两句吧...

相关阅读

    相关 Netty高性能的RPC框架

    Netty是一个开源的、高性能的、异步事件驱动的网络通信框架，支持多种协议和编码解码器，能够帮助开发人员快速构建高性能、可扩展的网络应用程序。它的主要优势包括： 1. 异步

    谁践踏了优雅谁践踏了优雅/ 0 赞/ 184 阅读

    相关 基于Netty实现简单的RPC框架

            Dubbo采用Netty作为基础通信组件，模仿Dubbo实现简单版的RPC框架。服务消费者和服务提供者约定接口和协议，用于远程调用和TCP请求验证。服务提供者作

    女爷i女爷i/ 0 赞/ 300 阅读