• 13.RPC的socket实现（阻塞式）与netty实现（非阻塞式）

    【README】

    • 1.本文总结了 RPC 的概念，包括定义，RPC实现，及其优缺点；
    • 2.本文po出了RPC的简单代码实现，总结自 B站《netty-尚硅谷》；
    • 3.本文部分内容总结自 ：What Is Remote Procedure Call (RPC)? Definition from SearchAppArchitecture
    • 4.本文还po出了 RPC 与 RMI 的区别；
    • 5.基于socket的rpc是阻塞式的，而基于netty的rpc是非阻塞式 ；

    【1】RPC 

    【1.1】RPC定义

    1）RPC定义： Remote Procedure Call， 远程过程调用协议；

    远程过程调用是一种软件通讯协议，基于该协议，一个程序可以通过网络调用另一台计算机上的服务而无需关心细节，就像调用本地系统的服务一样。

    【补充】RMI：Remote Method Invocation，远程方法调用

    • RMI 能够让jvm上的客户端像调用本地方法一样，调用远程jvm服务器上对象的方法；
    • 显然， RMI是 RPC 的 java版本的实现 ；

    2）常见的 RPC 框架有:

    • 比较知名的如阿里的Dubbo、google的gRPC、Go语言的rpcx、Apache的thrift， Spring 旗下的 Spring Cloud。

    【1.2】RPC的实现

    1）当触发远程过程调用，调用环境会被挂起，过程参数会通过网络传送到过程执行环境（过程参数可以理解为是类名，方法名，方法参数，方法参数类型等）。

    2）当服务器过程执行完成，结果会被回传到调用环境；只要过程调用返回则客户端立即恢复执行；

    3）RPC调用过程中，会经过如下步骤：

    • 步骤1：客户端调用客户端存根（代理）；该调用是本地过程调用，并把参数压入栈；
    • 步骤2：客户端代理封装过程参数到报文，并发送报文；把过程参数打包称为 参数编组；
    • 步骤3：客户端本地操作系统发送报文到远程服务器；
    • 步骤4：服务器操作系统把请求数据包送入 服务器存根（代理）；
    • 步骤5：服务器代理把参数解包，这个过程称为 参数解组；
    • 步骤6：当服务器过程调用完成，把结果返回给服务器代理；服务器代理把返回值封装到报文；服务器代理接着把报文传递给传输层；
    • 步骤7：服务器传输层发送响应报文到客户端传输层，接着回传给客户端代理；
    • 步骤8：客户端代理解组返回的参数，返回到调用者的执行点；

    4）RPC 调用流程图

    【图解】 在RPC中：

    • 1）客户端称为服务消费者；服务端称为服务提供者；
    • 2）编码是把对象转为字节数组（字节码），以便网络传输，即序列化；
    • 3）解码是把字节数组（字节）转为对象，以便业务逻辑处理，即反序列化，
    • 4）综上，RPC需要用到动态代理， 序列化与反序列化技术等；

     【1.3】RPC的优缺点

    1）优点：

    • 在高级语言中，有助于客户端通过传统过程调用与服务器通讯；
    • 能够被用于分布式环境，远程调用就像本地调用一样；
    • 支持面向进程与面向线程模型；
    • 对用户隐藏了内部报文传输机制；
    • 仅需极少的工作即可重写和重新开发代码；
    • 删除了许多协议层以提高性能；

    2）缺点：

    • 客户端与服务器各自使用不同执行环境，使用资源非常复杂，如文件。因此 RPC系统不适合传输大量数据；
    • RPC极易失败，因为它涉及到通讯系统，另一外机器和另一个处理过程；
    • RPC没有统一的标准，可以通过多种方式实现；
    • RPC 只是基于交互的，因此，它对硬件架构没有提供任何灵活性；

    【2】RPC的Socket实现（阻塞式）

    部分代码转自： 简单RPC之Socket实现_归田的博客-CSDN博客_rpc socket

    0）目录结构：

    【图解】RPC的实现代码包含3部分：

    1）公共协议：

    • 约定的报文格式（封装了过程调用参数，包括全限定类名，方法名，参数类型数组，参数值数组），以便于服务端（服务提供方）根据反射创建对应类与调用方法；
    • 约定的过程名（如方法名）：把方法名封装到接口；

    2）客户端（服务消费方）：

    • 通过代理模式创建服务的代理对象；调用方法，实际上调用的是代理对象方法，代理对象方法底层封装报文通过socket请求服务器获取调用结果；（客户端在整个调用过程是阻塞的，直到结果返回）

    3）服务端（服务提供方）：

    • 根据报文中的 类名通过反射创建服务实例，根据方法名，参数类型，参数列表通过反射调用方法，并将结果回写到socket；

    注意（报文的序列化与反序列化）：

    • 客户端把报文对象写出到 socket 前需要序列化为字节数组；服务器从socket读入字节数组后，需要反序列化为报文对象；

    【2.1】协议报文与过程调用方法定义：

    报文：

    1. /**
    2.  * @Description 报文（注意：要可序列化）
    3.  * @author xiao tang
    4.  * @version 1.0.0
    5.  * @createTime 2022年09月12日
    6.  */
    7. public class MyRpcMessage implements Externalizable {
    8.     /** 全限定类名 */
    9.     private String className;
    10.     /** 方法名 */
    11.     private String methodName;
    12.     /** 参数类型clazz数组 */
    13.     private Class[] parameterTypes;
    14.     /** 参数值 */
    15.     private Object[] parameterValues;
    16.  
    17.     /**
    18.      * @description 私有构造器
    19.      * @author xiao tang
    20.      * @date 2022/9/12
    21.      */
    22.     public MyRpcMessage() {
    23.     }
    24.  
    25.     /**
    26.      * @description 创建生成器
    27.      * @return 生成器
    28.      * @author xiao tang
    29.      * @date 2022/9/12
    30.      */
    31.     public static Builder builder() {
    32.         Builder builder =  new Builder();
    33.         return builder;
    34.     }
    35.  
    36.     public String getClassName() {
    37.         return className;
    38.     }
    39.  
    40.     public void setClassName(String className) {
    41.         this.className = className;
    42.     }
    43.  
    44.     public String getMethodName() {
    45.         return methodName;
    46.     }
    47.  
    48.     public void setMethodName(String methodName) {
    49.         this.methodName = methodName;
    50.     }
    51.  
    52.     public Class[] getParameterTypes() {
    53.         return parameterTypes;
    54.     }
    55.  
    56.     public void setParameterTypes(Class[] parameterTypes) {
    57.         this.parameterTypes = parameterTypes;
    58.     }
    59.  
    60.     public Object[] getParameterValues() {
    61.         return parameterValues;
    62.     }
    63.  
    64.     public void setParameterValues(Object[] parameterValues) {
    65.         this.parameterValues = parameterValues;
    66.     }
    67.  
    68.     @Override
    69.     public void writeExternal(ObjectOutput out) throws IOException {
    70.         out.writeUTF(className);
    71.         out.writeUTF(methodName);
    72.         out.writeObject(parameterTypes);
    73.         out.writeObject(parameterValues);
    74.     }
    75.  
    76.     @Override
    77.     public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
    78.         this.className = in.readUTF();
    79.         this.methodName = in.readUTF();
    80.         this.parameterTypes = (Class[]) in.readObject();
    81.         this.parameterValues = (Object[]) in.readObject();
    82.     }
    83.  
    84.     /**
    85.      * @description 生成器
    86.      * @author xiao tang
    87.      * @date 2022/9/12
    88.      */
    89.     public static class Builder {
    90.         /** rpc报文对象 */
    91.         private MyRpcMessage myRpcMessage;
    92.  
    93.         /**
    94.          * @description 私有构造器
    95.          * @author xiao tang
    96.          * @date 2022/9/12
    97.          */
    98.         private Builder() {
    99.             this.myRpcMessage = new MyRpcMessage();
    100.         }
    101.  
    102.         /**
    103.          * @description 设置全限定类名
    104.          * @return 生成器
    105.          * @author xiao tang
    106.          * @date 2022/9/12
    107.          */
    108.         public Builder className(String className) {
    109.             this.myRpcMessage.setClassName(className);
    110.             return this;
    111.         }
    112.  
    113.         /**
    114.          * @description 设置方法名
    115.          * @return 生成器
    116.          * @author xiao tang
    117.          * @date 2022/9/12
    118.          */
    119.         public Builder methodName(String methodName) {
    120.             this.myRpcMessage.setMethodName(methodName);
    121.             return this;
    122.         }
    123.  
    124.         /**
    125.          * @description 设置参数类型数组
    126.          * @return 生成器
    127.          * @author xiao tang
    128.          * @date 2022/9/12
    129.          */
    130.         public Builder parameterTypes(Class[] parameterTypes) {
    131.             this.myRpcMessage.setParameterTypes(parameterTypes);
    132.             return this;
    133.         }
    134.  
    135.         /**
    136.          * @description 设置参数值数组
    137.          * @return 生成器
    138.          * @author xiao tang
    139.          * @date 2022/9/12
    140.          */
    141.         public Builder parameterValues(Object[] parameterValues) {
    142.             this.myRpcMessage.setParameterValues(parameterValues);
    143.             return this;
    144.         }
    145.  
    146.         /**
    147.          * @description 获得rpc报文对象
    148.          * @return rpc报文对象
    149.          * @author xiao tang
    150.          * @date 2022/9/12
    151.          */
    152.         public MyRpcMessage build() {
    153.             return myRpcMessage;
    154.         }
    155.     }
    156.  
    157. }

    方法接口定义：

    1. /**
    2.  * @Description 过程调用接口
    3.  * @author xiao tang
    4.  * @version 1.0.0
    5.  * @createTime 2022年09月11日
    6.  */
    7. public interface IHelloRpc {
    8.  
    9.     String hello(String name);
    10. }

    【2.2】服务器：

    1. /**
    2.  * @Description rpc服务器
    3.  * @author xiao tang
    4.  * @version 1.0.0
    5.  * @createTime 2022年09月11日
    6.  */
    7. public class MyRpcSocketServer {
    8.     /** 线程池 */
    9.     private static ExecutorService THREAD_POOL = Executors.newCachedThreadPool();
    10.  
    11.     public static void main(String[] args) {
    12.         try {
    13.             startServer0(8089);
    14.         } catch (IOException e) {
    15.             e.printStackTrace();
    16.         } finally {
    17.             THREAD_POOL.shutdown();
    18.         }
    19.     }
    20.  
    21.     /**
    22.      * @description 开启服务器
    23.      * @param port 端口
    24.      * @author xiao tang
    25.      * @date 2022/9/11
    26.      */
    27.     private static void startServer0(int port) throws IOException {
    28.         ServerSocket serverSocket = new ServerSocket(port);
    29.         System.out.println("服务器启动成功.");
    30.         while(true) {
    31.             try {
    32.                 // 阻塞直到有客户端请求
    33.                 Socket socket = serverSocket.accept();
    34.                 // 处理器请求
    35.                 THREAD_POOL.execute(new BusiTask(socket));
    36.             } catch (Exception e) {
    37.                 e.printStackTrace();
    38.                 System.out.println("抛出异常");
    39.                 throw e;
    40.             }
    41.         }
    42.     }
    43.  
    44.     /**
    45.      * @description 业务作业，Runnable子类
    46.      * @author xiao tang
    47.      * @date 2022/9/11
    48.      */
    49.     static class BusiTask implements Runnable {
    50.         private Socket socket;
    51.         public BusiTask(Socket socket) {
    52.             this.socket = socket;
    53.         }
    54.         @Override
    55.         public void run() {
    56.             // 获得字节输入输出流
    57.             try (ObjectInputStream inputStream = new ObjectInputStream(socket.getInputStream());
    58.                     ObjectOutputStream outputMsg = new ObjectOutputStream(socket.getOutputStream());) {
    59.                 // 获取过程参数（报文），包括类名，方法名，方法参数类型，方法参数列表
    60.                 MyRpcMessage inputMsg = (MyRpcMessage)inputStream.readObject();
    61.                 // 通过反射创建对象
    62.                 Class clazz = Class.forName(inputMsg.getClassName());
    63.                 Object procedureObj = clazz.newInstance();
    64. //                // 通过反射创建方法
    65.                 Method method = clazz.getDeclaredMethod(inputMsg.getMethodName(), inputMsg.getParameterTypes());
    66. //                // 通过反射调用方法
    67.                 Object result = method.invoke(procedureObj, inputMsg.getParameterValues());
    68.                 // 把结果回写
    69.                 outputMsg.writeObject(result);
    70.             } catch (Exception e) {
    71.                 e.printStackTrace();
    72.             } finally {
    73.                 try {
    74.                     socket.close();
    75.                 } catch (IOException e) {
    76.                     e.printStackTrace();
    77.                 }
    78.             }
    79.         }
    80.     }
    81. }

    过程调用接口实现类：

    1. /**
    2.  * @Description 过程调用接口实现
    3.  * @author xiao tang
    4.  * @version 1.0.0
    5.  * @createTime 2022年09月11日
    6.  */
    7. public class IHelloRpcServerImpl implements IHelloRpc {
    8.     @Override
    9.     public String hello(String name) {
    10.         return "hello " + name + ", nice to meet you, i am a server named RPC.";
    11.     }
    12. }

    【2.3】客户端：

    1. /**
    2.  * @Description RPC存根（代理）工厂
    3.  * @author xiao tang
    4.  * @version 1.0.0
    5.  * @createTime 2022年09月11日
    6.  */
    7. public class MyRpcProxyFactory {
    8.  
    9.     /**
    10.      * @description 私有构造器
    11.      * @author xiao tang
    12.      * @date 2022/9/11
    13.      */
    14.     private MyRpcProxyFactory() {
    15.     }
    16.  
    17.     /**
    18.      * @description 创建代理对象
    19.      * @param targetClazz 目标对象clazz
    20.      * @param className 类名
    21.      * @return 代理对象
    22.      * @author xiao tang
    23.      * @date 2022/9/11
    24.      */
    25.     public static  T newProxy(Class targetClazz, String className) {
    26.         return (T) Proxy.newProxyInstance(
    27.                 targetClazz.getClassLoader()
    28.                 , new Class[]{targetClazz}
    29.                 , (Object proxy, Method method, Object[] args) ->{
    30.                         Socket socket = new Socket("localhost",8089);
    31.                         // 获取输出流
    32.                         try(ObjectOutputStream outputStream = new ObjectOutputStream(socket.getOutputStream());
    33.                             ObjectInputStream inputStream = new ObjectInputStream(socket.getInputStream())) {
    34.                             // 写出过程参数到服务器，包括类名，方法名，方法参数类型，方法参数列表
    35.                             MyRpcMessage message = MyRpcMessage.builder()
    36.                                                     .className(className)
    37.                                                     .methodName(method.getName())
    38.                                                     .parameterTypes(method.getParameterTypes())
    39.                                                     .parameterValues(args)
    40.                                                     .build();
    41.                             outputStream.writeObject(message);
    42.                             outputStream.flush();
    43.  
    44.                             // 获取输入流
    45.                             Object result = inputStream.readObject();
    46.                             if (result instanceof Throwable) {
    47.                                 throw (Throwable)result;
    48.                             }
    49.                             return result;
    50.                         } finally {
    51.                             socket.close();
    52.                         }
    53.                 });
    54.     }
    55. }

    测试用例入口：

    1. public class MyRpcSocketClient {
    2.  
    3.     public static void main(String[] args) {
    4.         IHelloRpc helloRpc = MyRpcProxyFactory.newProxy(IHelloRpc.class, "com.my.netty.rpc.socket.server.IHelloRpcServerImpl");
    5.         String result = helloRpc.hello("成都");
    6.         System.out.println(result);
    7.     }
    8. }

    【2.4】运行效果：

    hello 成都, nice to meet you, i am a server named RPC.

    【3】RPC的netty 实现

    1）背景： 基于socket的rpc实现时阻塞式的，吞吐量不高； 本文尝试使用netty 实现rpc ；

    2）netyt优势：

    • netty是非阻塞的，能够很好利用多核cpu的算力，增大系统吞吐量；
    • netty底层采用零拷贝，报文通讯性能更优；

    3）代码目录：

    【图解】 代码还是分为3部分（常量不算）：

    1）公共协议：

    • 把过程方法抽象为接口，客户端与服务器以该接口进行通讯；

    2）客户端：

    • 通过代理模式创建服务的代理对象；调用方法，实际上调用的是代理对象方法，代理对象方法底层封装报文（一个简单的字符串）通过 netty非阻塞式连接服务器，并把报文写出到channel，送入服务器（客户端在整个调用过程是非阻塞的）；
    • 当前线程睡眠直到服务端的响应报文回写到客户端；

    3）服务端：

    • 服务器在接收到客户端请求报文后，若协议报文头以 HelloServer#hello# 开头，则调用预设  HelloService.hello() 方法；
    • 调用hello() 方法完成后，把结果回写到netty的channel通道；
    • 补充：为了简化编程，基于netty的rpc实现没有使用反射来创建类和调用方法，而是写死 方法；当然了，像上文的socket版本的rpc一样，可以把类名，方法名，参数类型，参数值列表等过程参数封装到报文头，服务器可以根据上述报文头通过反射来创建类和调用方法；

    【3.1】公共协议（约定）

    1）公共接口；

    1. /**
    2.  * @Description 公共接口，服务提供方，服务消费方都需要
    3.  * @author xiao tang
    4.  * @version 1.0.0
    5.  * @createTime 2022年09月11日
    6.  */
    7. public interface HelloService {
    8.  
    9.     String hello(String msg);
    10. }

    常量：

    1. public class MyDubboConst {
    2.     /** 协议头 */
    3.     public static final String PROTOCOL_HEAD = "HelloServer#hello#";
    4. }

    【3.2】 客户端

    1）netty客户端代理：

    1. /**
    2.  * @Description 客户端-服务消费者
    3.  * @author xiao tang
    4.  * @version 1.0.0
    5.  * @createTime 2022年09月11日
    6.  */
    7. public class MyDubboNettyClient {
    8.     // 创建线程池
    9.     private static ExecutorService executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
    10.     // 客户端处理器
    11.     private static MyDubboNettyClientHandler clientHandler;
    12.  
    13.     private int counter = 0;
    14.  
    15.     // 编写方法，使用代理模式，获取代理对象
    16.     public Object getBean(final Class serviceClass, final String protocolHead) {
    17.         return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(),
    18.                 new Class[]{serviceClass}, (Object proxy, Method method, Object[] args)->{
    19.                     System.out.println("Object proxy, Method method, Object[] args 进入了，第 " + (++counter) + " 次进入");
    20.                     // 下面的代码， 客户端每调用一次hello，就会进入到该代码
    21.                     if (clientHandler == null) {
    22.                         initClientHandler(); // 初始化客户端处理器
    23.                     }
    24.                     // 设置要发送给服务器的信息
    25.                     // protocolHead 协议头部
    26.                     // args[0] 就是客户端调用 hello的参数
    27.                     System.out.println("clientHandler.setParameter");
    28.                     clientHandler.setParameter(protocolHead + args[0]);
    29.                     System.out.println("executorService.submit");
    30.                     return executorService.submit(clientHandler).get();
    31.                 });
    32.     }
    33.  
    34.     // 初始化客户端处理器
    35.     private static void initClientHandler() {
    36.         clientHandler = new MyDubboNettyClientHandler();
    37.         // 创建 EventLoopGroup
    38.         NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup();
    39.         Bootstrap bootstrap = new Bootstrap();
    40.         bootstrap.group(eventLoopGroup)
    41.                 .channel(NioSocketChannel.class)
    42.                 .option(ChannelOption.TCP_NODELAY, true)
    43.                 .handler(new ChannelInitializer() {
    44.                     @Override
    45.                     protected void initChannel(SocketChannel socketChannel) throws Exception {
    46.                         ChannelPipeline pipeline = socketChannel.pipeline();
    47.                         pipeline.addLast(new StringDecoder());
    48.                         pipeline.addLast(new StringEncoder());
    49.                         pipeline.addLast(clientHandler);
    50.                     }
    51.                 });
    52.         // 启动客户端
    53.         ChannelFuture channelFuture = null;
    54.         try {
    55.             channelFuture = bootstrap.connect("127.0.0.1", 8089).sync();
    56.         } catch (InterruptedException e) {
    57.             e.printStackTrace();
    58.         }
    59.         System.out.println("客户端启动成功");
    60.     }
    61. }

    2）netty客户端业务逻辑处理器

    1. /**
    2.  * @Description dubbo客户端处理器
    3.  * @author xiao tang
    4.  * @version 1.0.0
    5.  * @createTime 2022年09月11日
    6.  */
    7. public class MyDubboNettyClientHandler extends ChannelInboundHandlerAdapter implements Callable {
    8.  
    9.     private ChannelHandlerContext context; // 上下文
    10.     private String result; // 返回结果
    11.     private String parameter; // 客户端调用方法时，传入的参数
    12.  
    13.     // 与服务器连接创建成功后，就会被调用 (第1个被调用 )
    14.     @Override
    15.     public void channelActive(ChannelHandlerContext ctx) throws Exception {
    16.         System.out.println("MyDubboNettyClientHandler.channelActive");
    17.         context = ctx; // 因为在其他方法也会用到这个 context
    18.     }
    19.  
    20.     // 收到服务器的数据后，就会调用方法
    21.     @Override
    22.     public synchronized void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    23.         System.out.println("MyDubboNettyClientHandler.channelRead");
    24.         result = msg.toString();
    25.         notify(); // 唤醒等待的线程
    26.     }
    27.  
    28.     @Override
    29.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    30.         System.out.println("MyDubboNettyClientHandler.exceptionCaught");
    31.         ctx.close();
    32.     }
    33.  
    34.     // 被代理对象调用， 发送数据给服务器 -> 然后 wait -> 等待被唤醒  -> 返回结果 （第3个被调用）
    35.     @Override
    36.     public synchronized Object call() throws Exception {
    37.         System.out.println("发送数据到服务器， parameter = " + parameter);
    38.         context.writeAndFlush(parameter); // 发送数据给服务器
    39.         System.out.println("发送数据到服务成功");
    40.         wait(); // 等待直到服务器返回结果，服务器返回后调用 channelRead 方法来唤醒；
    41.         System.out.println("after wait().");
    42.         return result; // 服务提供方返回的结果
    43.     }
    44.     // 第2个被调用，设置参数
    45.     public void setParameter(String parameter) {
    46.         System.out.println("setParameter");
    47.         this.parameter = parameter;
    48.     }
    49. }

    3）客户端启动引导程序：

    1. public class MyDubboClientBootstrap {
    2.  
    3.     public static void main(String[] args) {
    4.         // 创建访问消费者-netty客户端
    5.         MyDubboNettyClient customer = new MyDubboNettyClient();
    6.         // 创建代理对象
    7.         HelloService helloService = (HelloService)customer.getBean(HelloService.class, MyDubboConst.PROTOCOL_HEAD);
    8.  
    9.         // 通过代理对象调用服务提供者的服务
    10.         String result = helloService.hello("hello服务器");
    11.         System.out.println("服务器响应结果 = " + result);
    12.     }
    13. }

    【3.3】服务器

    1）过程调用（方法调用）的具体实现

    1. /**
    2.  * @Description 服务提供方实现接口
    3.  * @author xiao tang
    4.  * @version 1.0.0
    5.  * @createTime 2022年09月11日
    6.  */
    7. public class HelloServiceImpl implements HelloService {
    8.     private int counter;
    9.     // 当有消费方调用该方法时，就返回一个字符串
    10.     @Override
    11.     public String hello(String msg) {
    12.         System.out.println("收到客户端消息 = " + msg);
    13.         // 根据msg 返回不同结果
    14.         if (msg != null) {
    15.             return "你好客户端，我收到你的消息 ["+msg+"]这是第" + (++counter) + "次收到消息";
    16.         } else {
    17.             return "你好客户端，我收到的消息为空" ;
    18.         }
    19.     }
    20. }

    2）netty服务器

    1. /**
    2.  * @Description 基于netty的 Dubbo 服务器
    3.  * @author xiao tang
    4.  * @version 1.0.0
    5.  * @createTime 2022年09月11日
    6.  */
    7. public class MyDubboNettyServer {
    8.  
    9.     public static void startServer(String hostName, int port) throws InterruptedException {
    10.         startServer0(hostName, port);
    11.     }
    12.  
    13.     private static void startServer0(String hostName, int port) throws InterruptedException {
    14.         EventLoopGroup bossGroup = new NioEventLoopGroup();
    15.         EventLoopGroup workerGroup = new NioEventLoopGroup();
    16.         try {
    17.             ServerBootstrap serverBootstrap = new ServerBootstrap();
    18.             serverBootstrap.group(bossGroup, workerGroup)
    19.                     .channel(NioServerSocketChannel.class)
    20.                     .childHandler(new ChannelInitializer() {
    21.                         @Override
    22.                         protected void initChannel(SocketChannel socketChannel) throws Exception {
    23.                             ChannelPipeline pipeline = socketChannel.pipeline();
    24.                             pipeline.addLast(new StringDecoder());
    25.                             pipeline.addLast(new StringEncoder());
    26.                             pipeline.addLast(new MyDubboNettyServerHandler()) ; // 自定义业务处理器
    27.                         }
    28.                     });
    29.             // 启动服务器
    30.             ChannelFuture channelFuture = serverBootstrap.bind(hostName, port).sync();
    31.             System.out.println("服务提供方启动成功，开始提供服务"); 
    32.             // 监听关闭
    33.             channelFuture.channel().closeFuture().sync();
    34.         } finally {
    35.             bossGroup.shutdownGracefully();
    36.             workerGroup.shutdownGracefully();
    37.         }
    38.     }
    39. }

    3）netty服务器业务处理器

    1. /**
    2.  * @Description 基于netty的 Dubbo 服务器处理器
    3.  * @author xiao tang
    4.  * @version 1.0.0
    5.  * @createTime 2022年09月11日
    6.  */
    7. public class MyDubboNettyServerHandler extends ChannelInboundHandlerAdapter {
    8.  
    9.     @Override
    10.     public void channelRead(ChannelHandlerContext ctx, Object msg) {
    11.         // 获取客户端发送的消息 并 调用公共接口服务
    12.         System.out.println("接收的msg = " + msg);
    13.         // 客户端在调用服务器的api时，我们需要定义一个协议
    14.         // 如 msg 以 "HelloServer#hello#" 开头，才调用api
    15.         String msgString = msg.toString();
    16.         if (msgString.startsWith(MyDubboConst.PROTOCOL_HEAD)) {
    17.             String result = new HelloServiceImpl().hello(msgString.substring(msgString.lastIndexOf("#")+1));
    18.             ctx.writeAndFlush(result);
    19.         } else {
    20.             ctx.writeAndFlush("协议头非法");
    21.         }
    22.     }
    23.  
    24.     @Override
    25.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    26.         cause.printStackTrace();
    27.         ctx.close();
    28.     }
    29. }

    4）服务器启动引导程序：

    1. public class MyDubboServerBootstrap {
    2.     public static void main(String[] args) throws InterruptedException {
    3.         // 代码代填
    4.         MyDubboNettyServer.startServer("127.0.0.1", 8089);
    5.     }
    6. }

    【3.4】运行效果：

    1）服务器日志：

    1. 服务提供方启动成功，开始提供服务
    2. [DEBUG] Recycler - -Dio.netty.recycler.maxCapacityPerThread: 32768
    3. [DEBUG] Recycler - -Dio.netty.recycler.maxSharedCapacityFactor: 2
    4. [DEBUG] Recycler - -Dio.netty.recycler.linkCapacity: 16
    5. [DEBUG] Recycler - -Dio.netty.recycler.ratio: 8
    6. [DEBUG] AbstractByteBuf - -Dio.netty.buffer.bytebuf.checkAccessible: true
    7. [DEBUG] ResourceLeakDetectorFactory$DefaultResourceLeakDetectorFactory - Loaded default ResourceLeakDetector: io.netty.util.ResourceLeakDetector@e62edfd
    8. 接收的msg = HelloServer#hello#hello服务器
    9. 收到客户端消息 = hello服务器

    2）客户端日志：

    1. 客户端启动成功
    2. clientHandler.setParameter
    3. MyDubboNettyClientHandler.channelActive
    4. setParameter
    5. executorService.submit
    6. 发送数据到服务器， parameter = HelloServer#hello#hello服务器
    7. [DEBUG] Recycler - -Dio.netty.recycler.maxCapacityPerThread: 32768
    8. [DEBUG] Recycler - -Dio.netty.recycler.maxSharedCapacityFactor: 2
    9. [DEBUG] Recycler - -Dio.netty.recycler.linkCapacity: 16
    10. [DEBUG] Recycler - -Dio.netty.recycler.ratio: 8
    11. 发送数据到服务成功
    12. [DEBUG] AbstractByteBuf - -Dio.netty.buffer.bytebuf.checkAccessible: true
    13. [DEBUG] ResourceLeakDetectorFactory$DefaultResourceLeakDetectorFactory - Loaded default ResourceLeakDetector: io.netty.util.ResourceLeakDetector@89e890f
    14. MyDubboNettyClientHandler.channelRead
    15. after wait().
    16. 服务器响应结果 = 你好客户端，我收到你的消息 [hello服务器]这是第1次收到消息

     

  • 相关阅读:
    HCIE Routing&Switching之MPLS LDP理论
    CLUE模型构建方法、模型验证及土地利用变化情景预测实践技术
    mysql 闪回表工具
    1201. Ugly Number III && 264. Ugly Number ll
    [附源码]计算机毕业设计springboot大学生考勤管理系统论文
    require、loadfile、dofile、load、loadstring
    vscode连接服务器
    LabVIEW中图像显示错误
    kubelet gc 源码分析
    Hopcroft–Karp algorithm
  • 原文地址：https://blog.csdn.net/PacosonSWJTU/article/details/126808608