由浅入深Netty简易实现RPC框架

叁歲伎倆 2024-03-17 09:13 180阅读 0赞

目录

    • 1 准备工作
    • 2 服务器 handler
    • 3 客户端代码第一版
    • 4 客户端 handler 第一版
    • 5 客户端代码 第二版
    • 6 客户端 handler 第二版

1 准备工作

在这里插入图片描述

这些代码可以认为是现成的，无需从头编写练习

为了简化起见，在原来聊天项目的基础上新增 Rpc 请求和响应消息

  1. @Data
  2. public abstract class Message implements Serializable {
  5.     // 省略旧的代码
  7.     public static final int RPC_MESSAGE_TYPE_REQUEST = 101;
  8.     public static final int  RPC_MESSAGE_TYPE_RESPONSE = 102;
  10.     static {
  12.         // ...
  13.         messageClasses.put(RPC_MESSAGE_TYPE_REQUEST, RpcRequestMessage.class);
  14.         messageClasses.put(RPC_MESSAGE_TYPE_RESPONSE, RpcResponseMessage.class);
  15.     }
  17. }

请求消息

  1. @Getter
  2. @ToString(callSuper = true)
  3. public class RpcRequestMessage extends Message {
  6.     /**
  7.      * 调用的接口全限定名，服务端根据它找到实现
  8.      */
  9.     private String interfaceName;
  10.     /**
  11.      * 调用接口中的方法名
  12.      */
  13.     private String methodName;
  14.     /**
  15.      * 方法返回类型
  16.      */
  17.     private Class<?> returnType;
  18.     /**
  19.      * 方法参数类型数组
  20.      */
  21.     private Class[] parameterTypes;
  22.     /**
  23.      * 方法参数值数组
  24.      */
  25.     private Object[] parameterValue;
  27.     public RpcRequestMessage(int sequenceId, String interfaceName, String methodName, Class<?> returnType, Class[] parameterTypes, Object[] parameterValue) {
  29.         super.setSequenceId(sequenceId);
  30.         this.interfaceName = interfaceName;
  31.         this.methodName = methodName;
  32.         this.returnType = returnType;
  33.         this.parameterTypes = parameterTypes;
  34.         this.parameterValue = parameterValue;
  35.     }
  37.     @Override
  38.     public int getMessageType() {
  40.         return RPC_MESSAGE_TYPE_REQUEST;
  41.     }
  42. }

响应消息

  1. @Data
  2. @ToString(callSuper = true)
  3. public class RpcResponseMessage extends Message {
  5.     /**
  6.      * 返回值
  7.      */
  8.     private Object returnValue;
  9.     /**
  10.      * 异常值
  11.      */
  12.     private Exception exceptionValue;
  14.     @Override
  15.     public int getMessageType() {
  17.         return RPC_MESSAGE_TYPE_RESPONSE;
  18.     }
  19. }

服务器架子

  1. @Slf4j
  2. public class RpcServer {
  4.     public static void main(String[] args) {
  6.         NioEventLoopGroup boss = new NioEventLoopGroup();
  7.         NioEventLoopGroup worker = new NioEventLoopGroup();
  8.         LoggingHandler LOGGING_HANDLER = new LoggingHandler(LogLevel.DEBUG);
  9.         MessageCodecSharable MESSAGE_CODEC = new MessageCodecSharable();
  11.         // rpc 请求消息处理器，待实现
  12.         RpcRequestMessageHandler RPC_HANDLER = new RpcRequestMessageHandler();
  13.         try {
  15.             ServerBootstrap serverBootstrap = new ServerBootstrap();
  16.             serverBootstrap.channel(NioServerSocketChannel.class);
  17.             serverBootstrap.group(boss, worker);
  18.             serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
  20.                 @Override
  21.                 protected void initChannel(SocketChannel ch) throws Exception {
  23.                     ch.pipeline().addLast(new ProcotolFrameDecoder());
  24.                     ch.pipeline().addLast(LOGGING_HANDLER);
  25.                     ch.pipeline().addLast(MESSAGE_CODEC);
  26.                     ch.pipeline().addLast(RPC_HANDLER);
  27.                 }
  28.             });
  29.             Channel channel = serverBootstrap.bind(8080).sync().channel();
  30.             channel.closeFuture().sync();
  31.         } catch (InterruptedException e) {
  33.             log.error("server error", e);
  34.         } finally {
  36.             boss.shutdownGracefully();
  37.             worker.shutdownGracefully();
  38.         }
  39.     }
  40. }

客户端架子

  1. public class RpcClient {
  3.     public static void main(String[] args) {
  5.         NioEventLoopGroup group = new NioEventLoopGroup();
  6.         LoggingHandler LOGGING_HANDLER = new LoggingHandler(LogLevel.DEBUG);
  7.         MessageCodecSharable MESSAGE_CODEC = new MessageCodecSharable();
  9.         // rpc 响应消息处理器，待实现
  10.         RpcResponseMessageHandler RPC_HANDLER = new RpcResponseMessageHandler();
  11.         try {
  13.             Bootstrap bootstrap = new Bootstrap();
  14.             bootstrap.channel(NioSocketChannel.class);
  15.             bootstrap.group(group);
  16.             bootstrap.handler(new ChannelInitializer<SocketChannel>() {
  18.                 @Override
  19.                 protected void initChannel(SocketChannel ch) throws Exception {
  21.                     ch.pipeline().addLast(new ProcotolFrameDecoder());
  22.                     ch.pipeline().addLast(LOGGING_HANDLER);
  23.                     ch.pipeline().addLast(MESSAGE_CODEC);
  24.                     ch.pipeline().addLast(RPC_HANDLER);
  25.                 }
  26.             });
  27.             Channel channel = bootstrap.connect("localhost", 8080).sync().channel();
  28.             channel.closeFuture().sync();
  29.         } catch (Exception e) {
  31.             log.error("client error", e);
  32.         } finally {
  34.             group.shutdownGracefully();
  35.         }
  36.     }
  37. }

服务器端的 service 获取

  1. public class ServicesFactory {
  4.     static Properties properties;
  5.     static Map<Class<?>, Object> map = new ConcurrentHashMap<>();
  7.     static {
  9.         try (InputStream in = Config.class.getResourceAsStream("/application.properties")) {
  11.             properties = new Properties();
  12.             properties.load(in);
  13.             Set<String> names = properties.stringPropertyNames();
  14.             for (String name : names) {
  16.                 if (name.endsWith("Service")) {
  18.                     Class<?> interfaceClass = Class.forName(name);
  19.                     Class<?> instanceClass = Class.forName(properties.getProperty(name));
  20.                     map.put(interfaceClass, instanceClass.newInstance());
  21.                 }
  22.             }
  23.         } catch (IOException | ClassNotFoundException | InstantiationException | IllegalAccessException e) {
  25.             throw new ExceptionInInitializerError(e);
  26.         }
  27.     }
  29.     public static <T> T getService(Class<T> interfaceClass) {
  31.         return (T) map.get(interfaceClass);
  32.     }
  33. }

相关配置 application.properties

  1. serializer.algorithm=Json
  2. cn.itcast.server.service.HelloService=cn.itcast.server.service.HelloServiceImpl

2 服务器 handler

  1. @Slf4j
  2. @ChannelHandler.Sharable
  3. public class RpcRequestMessageHandler extends SimpleChannelInboundHandler<RpcRequestMessage> {
  6.     @Override
  7.     protected void channelRead0(ChannelHandlerContext ctx, RpcRequestMessage message) {
  9.         RpcResponseMessage response = new RpcResponseMessage();
  10.         response.setSequenceId(message.getSequenceId());
  11.         try {
  13.             // 获取真正的实现对象
  14.             HelloService service = (HelloService)
  15.                     ServicesFactory.getService(Class.forName(message.getInterfaceName()));
  17.             // 获取要调用的方法
  18.             Method method = service.getClass().getMethod(message.getMethodName(), message.getParameterTypes());
  20.             // 调用方法
  21.             Object invoke = method.invoke(service, message.getParameterValue());
  22.             // 调用成功
  23.             response.setReturnValue(invoke);
  24.         } catch (Exception e) {
  26.             e.printStackTrace();
  27.             // 调用异常
  28.             response.setExceptionValue(e);
  29.         }
  30.         // 返回结果
  31.         ctx.writeAndFlush(response);
  32.     }
  33. }

3 客户端代码第一版

只发消息

  1. @Slf4j
  2. public class RpcClient {
  4.     public static void main(String[] args) {
  6.         NioEventLoopGroup group = new NioEventLoopGroup();
  7.         LoggingHandler LOGGING_HANDLER = new LoggingHandler(LogLevel.DEBUG);
  8.         MessageCodecSharable MESSAGE_CODEC = new MessageCodecSharable();
  9.         RpcResponseMessageHandler RPC_HANDLER = new RpcResponseMessageHandler();
  10.         try {
  12.             Bootstrap bootstrap = new Bootstrap();
  13.             bootstrap.channel(NioSocketChannel.class);
  14.             bootstrap.group(group);
  15.             bootstrap.handler(new ChannelInitializer<SocketChannel>() {
  17.                 @Override
  18.                 protected void initChannel(SocketChannel ch) throws Exception {
  20.                     ch.pipeline().addLast(new ProcotolFrameDecoder());
  21.                     ch.pipeline().addLast(LOGGING_HANDLER);
  22.                     ch.pipeline().addLast(MESSAGE_CODEC);
  23.                     ch.pipeline().addLast(RPC_HANDLER);
  24.                 }
  25.             });
  26.             Channel channel = bootstrap.connect("localhost", 8080).sync().channel();
  28.             ChannelFuture future = channel.writeAndFlush(new RpcRequestMessage(
  29.                     1,
  30.                     "cn.itcast.server.service.HelloService",
  31.                     "sayHello",
  32.                     String.class,
  33.                     new Class[]{
  34.     String.class},
  35.                     new Object[]{
  36.     "张三"}
  37.             )).addListener(promise -> {
  39.                 if (!promise.isSuccess()) {
  41.                     Throwable cause = promise.cause();
  42.                     log.error("error", cause);
  43.                 }
  44.             });
  46.             channel.closeFuture().sync();
  47.         } catch (Exception e) {
  49.             log.error("client error", e);
  50.         } finally {
  52.             group.shutdownGracefully();
  53.         }
  54.     }
  55. }

4 客户端 handler 第一版

  1. @Slf4j
  2. @ChannelHandler.Sharable
  3. public class RpcResponseMessageHandler extends SimpleChannelInboundHandler<RpcResponseMessage> {
  5.     @Override
  6.     protected void channelRead0(ChannelHandlerContext ctx, RpcResponseMessage msg) throws Exception {
  8.         log.debug("{}", msg);
  9.     }
  10. }

5 客户端代码 第二版

包括 channel 管理，代理，接收结果

  1. @Slf4j
  2. public class RpcClientManager {
  6.     public static void main(String[] args) {
  8.         HelloService service = getProxyService(HelloService.class);
  9.         System.out.println(service.sayHello("zhangsan"));
  10. //        System.out.println(service.sayHello("lisi"));
  11. //        System.out.println(service.sayHello("wangwu"));
  12.     }
  14.     // 创建代理类
  15.     public static <T> T getProxyService(Class<T> serviceClass) {
  17.         ClassLoader loader = serviceClass.getClassLoader();
  18.         Class<?>[] interfaces = new Class[]{
  19.     serviceClass};
  20.         //                                                            sayHello  "张三"
  21.         Object o = Proxy.newProxyInstance(loader, interfaces, (proxy, method, args) -> {
  23.             // 1. 将方法调用转换为 消息对象
  24.             int sequenceId = SequenceIdGenerator.nextId();
  25.             RpcRequestMessage msg = new RpcRequestMessage(
  26.                     sequenceId,
  27.                     serviceClass.getName(),
  28.                     method.getName(),
  29.                     method.getReturnType(),
  30.                     method.getParameterTypes(),
  31.                     args
  32.             );
  33.             // 2. 将消息对象发送出去
  34.             getChannel().writeAndFlush(msg);
  36.             // 3. 准备一个空 Promise 对象，来接收结果             指定 promise 对象异步接收结果线程
  37.             DefaultPromise<Object> promise = new DefaultPromise<>(getChannel().eventLoop());
  38.             RpcResponseMessageHandler.PROMISES.put(sequenceId, promise);
  40. //            promise.addListener(future -> {
  42. //                // 线程
  43. //            });
  45.             // 4. 等待 promise 结果
  46.             promise.await();
  47.             if(promise.isSuccess()) {
  49.                 // 调用正常
  50.                 return promise.getNow();
  51.             } else {
  53.                 // 调用失败
  54.                 throw new RuntimeException(promise.cause());
  55.             }
  56.         });
  57.         return (T) o;
  58.     }
  60.     private static Channel channel = null;
  61.     private static final Object LOCK = new Object();
  63.     // 获取唯一的 channel 对象
  64.     public static Channel getChannel() {
  66.         if (channel != null) {
  68.             return channel;
  69.         }
  70.         synchronized (LOCK) {
  71.      //  t2
  72.             if (channel != null) {
  73.      // t1
  74.                 return channel;
  75.             }
  76.             initChannel();
  77.             return channel;
  78.         }
  79.     }
  81.     // 初始化 channel 方法
  82.     private static void initChannel() {
  84.         NioEventLoopGroup group = new NioEventLoopGroup();
  85.         LoggingHandler LOGGING_HANDLER = new LoggingHandler(LogLevel.DEBUG);
  86.         MessageCodecSharable MESSAGE_CODEC = new MessageCodecSharable();
  87.         RpcResponseMessageHandler RPC_HANDLER = new RpcResponseMessageHandler();
  88.         Bootstrap bootstrap = new Bootstrap();
  89.         bootstrap.channel(NioSocketChannel.class);
  90.         bootstrap.group(group);
  91.         bootstrap.handler(new ChannelInitializer<SocketChannel>() {
  93.             @Override
  94.             protected void initChannel(SocketChannel ch) throws Exception {
  96.                 ch.pipeline().addLast(new ProcotolFrameDecoder());
  97.                 ch.pipeline().addLast(LOGGING_HANDLER);
  98.                 ch.pipeline().addLast(MESSAGE_CODEC);
  99.                 ch.pipeline().addLast(RPC_HANDLER);
  100.             }
  101.         });
  102.         try {
  104.             channel = bootstrap.connect("localhost", 8080).sync().channel();
  105.             channel.closeFuture().addListener(future -> {
  107.                 group.shutdownGracefully();
  108.             });
  109.         } catch (Exception e) {
  111.             log.error("client error", e);
  112.         }
  113.     }
  114. }

6 客户端 handler 第二版

  1. @Slf4j
  2. @ChannelHandler.Sharable
  3. public class RpcResponseMessageHandler extends SimpleChannelInboundHandler<RpcResponseMessage> {
  6.     //                       序号      用来接收结果的 promise 对象
  7.     public static final Map<Integer, Promise<Object>> PROMISES = new ConcurrentHashMap<>();
  9.     @Override
  11.     protected void channelRead0(ChannelHandlerContext ctx, RpcResponseMessage msg) throws Exception {
  13.         log.debug("{}", msg);
  14.         // 拿到空的 promise
  15.         Promise<Object> promise = PROMISES.remove(msg.getSequenceId());
  16.         if (promise != null) {
  18.             Object returnValue = msg.getReturnValue();
  19.             Exception exceptionValue = msg.getExceptionValue();
  20.             if(exceptionValue != null) {
  22.                 promise.setFailure(exceptionValue);
  23.             } else {
  25.                 promise.setSuccess(returnValue);
  26.             }
  27.         }
  28.     }
  29. }

发表评论

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

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

相关阅读

    相关 Netty高性能的RPC框架

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

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

    相关 手写RPC框架(netty+zookeeper)

    　　RPC是什么？远程过程调用，过程就是业务处理、计算任务，像调用本地方法一样调用远程的过程。 　　RMI和RPC的区别是什么？RMI是远程方法调用，是oop领域中RPC的一

    Bertha 。Bertha 。/ 0 赞/ 255 阅读

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

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

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