自定义RPC框架

分布式架构网络通信

分布式的基础问题是远程服务是怎么通讯的。

java 领域有很多可实现远程通讯的技术，例如：RMI 、Hessian 、SOAP 、ESB 和 JMS 等。

远程通讯技术

RMI

JDK的RMI文档：https://docs.oracle.com/javase/8/docs/technotes/guides/rmi/

Hessian

Hessian官网：http://hessian.caucho.com/

SOAP

SOAP：https://zh.wikipedia.org/wiki/%E7%AE%80%E5%8D%95%E5%AF%B9%E8%B1%A1%E8%AE%BF%E9%97%AE%E5%8D%8F%E8%AE%AE

• WSDL WS-* are language-agnostic.

• JAX-WS are Java standard to build web service.

• Apache CXF and Apache Axis 2 are two implementations of JAX-WS. They also offer JAX-RS implementations so that you can build Restful services.

• CXF has better integration with Spring, and Camel(camel-cxf). And Axis 2 seems not have a active release.

• https://github.com/eclipse-ee4j/jersey

• https://cxf.apache.org/index.html

• https://github.com/apache/axis-axis2-java-core

ESB

ESB：https://zh.wikipedia.org/wiki/%E4%BC%81%E4%B8%9A%E6%9C%8D%E5%8A%A1%E6%80%BB%E7%BA%BF

• Spring Integration：https://spring.io/projects/spring-integration

• Mule ESB：https://www.mulesoft.com/resources/esb/what-mule-esb

• Apache Camel：https://camel.apache.org/

JMS

JMS：https://www.oracle.com/java/technologies/java-message-service.html

https://spring.io/guides/gs/messaging-jms/

• Spring JMS is the Spring abstraction over the JMS API.
• JMS is just an API; you need a physical broker to actually do messaging.
• ActiveMQ is not a framework, it is an open-source JMS broker that supports the actual persistence and delivery of messages.
• Spring JMS can be used with any JMS broker, including ActiveMQ. Each broker provides its own JMS API client jar.
• RabbitMQ is not a native JMS broker; its native protocol is AMQP 0.9.1; it does, however, provide a JMS API client that can be used with Spring JMS, but Spring AMQP is the preferred library for talking to RabbitMQ because it provides much more functionality than is available over JMS.
• There are lots of examples for using Spring JMS on the internet.
• The simplest way to get started is with Spring Boot and @JmsListener.

基本原理

从网络通信的底层去看，通信要做的事情就是把流从一台计算机传输到另外一台计算机。使用传输协议和网络IO实现，传输协议比较知名的如 TCP 、UDP 等。

TCP 和 UDP 都是在基于 socket 的概念上为某类应用场景扩展出来的传输层协议。

网络 IO 主要由 bio 、nio、aio，所有的分布式都是基于这个原理实现的。

什么是RPC

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

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

下面展示的是本地调用和远程调用的示例：

例如有A服务器部署了一个应用，B服务器也部署了一个应用，现在A服务器上的应用想要调用B服务器上的应用的方法，由于，两个应用不在同一个服务器，因此不在同一个内存空间，无法实现直接调用，需要通过网络来表达调用的语义和传达调用的数据。

<img src="media/16461457884893/16467573463113.jpg" style="zoom:50%;" />

<img src="media/16461457884893/16467573463123.jpg" style="zoom:50%;" />

RPC架构

一个完整的RPC架构包含四个完整的组件，分表是Client，Client Stub，Server和Server Stub。Stub可以理解为存根。

• 客户端（Client），服务的调用方。
• 客户端存根（Client Stub），存放服务端的地址消息，将客户端的请求打包成网络消息，通过网络远程发送给服务方。
• 服务端（Server），真正的服务端提供者。
• 服务端存根（Sever Stub），接收客户端发送过来的消息，将消息解包，并调用本地方法。

<img src="media/16461457884893/16467573463136.jpg" style="zoom:50%;" />

<img src="media/16461457884893/16467573463152.jpg" style="zoom:50%;" />

1、客户端以本地方式调用服务

2、客户端存根接收到调用之后，将方法参数组装成能进行网络传输的消息体，消息体序列化为二进制

3、客户端通过socket将消息发送到服务端

4、服务端存根接收到消息之后进行解码，将消息对象反序列化

5、服务端存根根据解码结果调用本地服务

6、服务处理

7、本地服务执行并将结果返回给服务端存根

8、服务端存根将返回结果打包成消息，将结果消息对象序列化

9、服务端通过socket将消息发送到客户端

10、客户端存根接收到消息并进行解码，将消息对象反序列化

11、客户端得到最终结果

RPC的目标是只保留1、6、11，将其他的细节全部封装起来。

注意：不管是什么类型的数据，在输出过程中都要转换成二进制流，而接收方需要将二进制流恢复为对象。

Java中常见的RPC框架有Hessian、gRPC、Dubbo等，核心模块都是通讯和序列化

https://github.com/grpc/grpc-java

RMI

Java的RMI指的是 Remote Method Invocation，一种实现远程过程调用（RPC）的API，能直接传输序列化后的Java对象。它的实现依赖于JVM，因此它能支撑一个JVM到另外一个JVM的调用。

<img src="media/16461457884893/16467573463168.jpg" style="zoom:50%;" />

1、客户端从远程服务器的注册表中查询并获取远程对象的引用。

2、桩对象与远程对象有相同的接口和方法列表，当客户端调用远程对象时候，实际上是由桩对象代理完成。

3、远程引用层将桩的本地引用转换为服务器上对象的远程引用，再将调用层传递给传输层，由传输层通过TCP协议发起调用。

4、在服务端，传输层监听入站链接，收到客户端的远程调用之后，将引用转发到上层的远程引用层；

服务端的远程引用层将客户端发送的远程引用转换为本地虚拟机的引用，再将请求传递给骨架；

骨架读取参数，将请求传递给服务器，由服务器进行实际的方法调用。

5、如果远程方法调用之后有返回值，服务器将结果沿着 "骨架->远程引用层->传输层" 向下传递。

6、客户端的传输层接收到返回值之后，又沿着 "传输层->远程引用层->桩" 向上传递，并最终将结果传递给客户端程序。

RMI实例需求分析

1、服务端提供根据ID查询用户的方法

2、客户端调用服务端方法，并返回用户对象

3、要求使用RMI进行远程通讯

服务端实现

/**
 * RMI服务端
 *
 * @name: RMIServer
 * @author: terwer
 * @date: 2022-03-06 02:01
 **/
public class RMIServer {
    public static void main(String[] args) {
        try {
            // 1.注册Registry实例，绑定端口
            Registry registry = LocateRegistry.createRegistry(9998);
            // 2.创建远程对象
            IUserService userService = new UserServiceImpl();
            // 3.将远程对象注册到RMI服务器（既服务端注册表）
            registry.rebind("userService", userService);

            System.out.println("RMI服务端启动成功");
        } catch (RemoteException e) {
            e.printStackTrace();
        }
    }
}

客户端实现

/**
 * RMI客户端
 *
 * @name: RMIClient
 * @author: terwer
 * @date: 2022-03-06 19:25
 **/
public class RMIClient {
    public static void main(String[] args) throws RemoteException, NotBoundException {
        // 1、获取Registry实例
        Registry registry = LocateRegistry.getRegistry("127.0.0.1", 9998);
        // 2、通过Registry查找远程对象
        IUserService userService = (IUserService) registry.lookup("userService");
        User user = userService.getUserById(1);
        System.out.println("userName = " + user.getName());
    }
}

效果预览

<img src="media/16461457884893/16467573463195.png" alt="image-20220306193226744" style="zoom:50%;" />

<img src="media/16461457884893/16467573463236.png" alt="image-20220306193244530" style="zoom:50%;" />

基于Netty实现RPC框架

Dubbo 底层使用 Netty 作为网络通讯框架，要求使用 Netty 实现一个简单的 RPC 框架，消费者和提供者约定协议和接口，消费者远程调用提供者的服务。

1、创建一个接口，定义抽象方法，用于消费者和提供者之间的约定。

2、创建一个提供者，该类需要监听消费者的请求，并按照约定返回数据。

3、创建一个消费者，该类需要透明的调用自己不存在的方法，内部需要使用 Netty 进行数据通信。

4、提供者与消费者传输数据使用json字符串格式。

5、提供者使用 Netty 集成 Spring Boot 环境。

案例：客户端调用服务端，利用ID查询User对象的方法

需求分析

<img src="media/16461457884893/16467573463305.png" alt="image-20220306234352362" style="zoom:50%;" />

具体实现

需要分成三个子项目

.
├── custom-rpc-api
├── custom-rpc-consumer
├── custom-rpc-provider
└── pom.xml

主项目

主项目的 pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xmlns="http://maven.apache.org/POM/4.0.0"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>com.terewrgreen</groupId>
    <artifactId>custom-rpc</artifactId>
    <packaging>pom</packaging>
    <version>1.0.0</version>

    <modules>
        <module>custom-rpc-api</module>
        <module>custom-rpc-provider</module>
        <module>custom-rpc-consumer</module>
    </modules>

    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.6.3</version>
    </parent>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <curator.version>4.3.0</curator.version>
    </properties>

    <dependencies>
        <!--netty依赖 -->
        <dependency>
            <groupId>io.netty</groupId>
            <artifactId>netty-all</artifactId>
        </dependency>
        <!--json依赖 -->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.79</version>
        </dependency>
        <!--lombok依赖 -->
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>
    </dependencies>
</project>

custom-rpc-api

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>custom-rpc</artifactId>
        <groupId>com.terewrgreen</groupId>
        <version>1.0.0</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>custom-rpc-api</artifactId>

    <name>custom-rpc-api</name>
    <url>http://www.terwergreen.com</url>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <maven.compiler.source>1.8</maven.compiler.source>
        <maven.compiler.target>1.8</maven.compiler.target>
    </properties>

    <dependencies>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.13.1</version>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <build>
        <pluginManagement>
        </pluginManagement>
    </build>
</project>

custom-rpc-consumer

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>custom-rpc</artifactId>
        <groupId>com.terewrgreen</groupId>
        <version>1.0.0</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>custom-rpc-consumer</artifactId>

    <name>custom-rpc-consumer</name>
    <url>http://www.terwergreen.com</url>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <maven.compiler.source>1.8</maven.compiler.source>
        <maven.compiler.target>1.8</maven.compiler.target>
    </properties>

    <dependencies>
        <dependency>
            <groupId>com.terewrgreen</groupId>
            <artifactId>custom-rpc-api</artifactId>
            <version>1.0.0</version>
        </dependency>

        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.13.1</version>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <build>
        <pluginManagement>
        </pluginManagement>
    </build>
</project>

RpcClient

/**
 * Rpc客户端
 * 1、连接netty服务端
 * 2、提供给调用者关闭资源的方法
 * 3、提供消息发送的方法
 *
 * @name: RpcClient
 * @author: terwer
 * @date: 2022-03-13 21:04
 **/
public class RpcClient {

    private NioEventLoopGroup group;
    private Channel channel;

    private String ip;
    private int port;

    private RpcClientHandler rpcClientHandler = new RpcClientHandler();
    private ExecutorService executorService = Executors.newCachedThreadPool();

    public RpcClient(String ip, int port) {
        this.ip = ip;
        this.port = port;
        initClient();
    }

    /**
     * 初始化客户端，连接netty服务端
     */
    public void initClient() {

        try {
            // 创建线程组
            group = new NioEventLoopGroup();
            //  创建启动助手
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.SO_KEEPALIVE, true)
                    .option(ChannelOption.SO_TIMEOUT, 3000)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel channel) throws Exception {
                            ChannelPipeline pipeline = channel.pipeline();
                            // String 编解码器
                            pipeline.addLast(new StringDecoder());
                            pipeline.addLast(new StringEncoder());
                            // 客户端处理类
                            pipeline.addLast(rpcClientHandler);
                        }
                    });
            channel = bootstrap.connect(ip, port).sync().channel();
            System.out.println("===========客户端启动成功==========");
        } catch (Exception e) {
            if (channel != null) {
                channel.close();
                System.out.println("客户端关闭channel");
            }
            if (group != null) {
                group.shutdownGracefully();
                System.out.println("客户端关闭group");
            }
            e.printStackTrace();
        }
    }

    public void close(){
        if (channel != null) {
            channel.close();
            System.out.println("外部调用客户端关闭channel");
        }
        if (group != null) {
            group.shutdownGracefully();
            System.out.println("外部调用客户端关闭group");
        }
    }

    public Object send(String msg) throws ExecutionException, InterruptedException {
        rpcClientHandler.setRequestMessage(msg);
        Future future = executorService.submit(rpcClientHandler);
        return future.get();
    }
}

RpcClienthandler

/**
 * 客户端处理类
 * 1、发送消息
 * 2、接收消息
 *
 * @name: RpcClientHandler
 * @author: terwer
 * @date: 2022-03-13 23:01
 **/
public class RpcClientHandler extends SimpleChannelInboundHandler implements Callable {

    private ChannelHandlerContext ctx;
    // 消息
    private String requestMessage;
    private String responseMessage;

    public String getRequestMessage() {
        return requestMessage;
    }

    public void setRequestMessage(String requestMessage) {
        this.requestMessage = requestMessage;
    }

    /**
     * 通道读取就绪事件
     *
     * @param channelHandlerContext
     * @param msg
     * @throws Exception
     */
    @Override
    protected synchronized void channelRead0(ChannelHandlerContext channelHandlerContext, Object msg) throws Exception {
        responseMessage = (String) msg;
        // 唤醒等待线程
        notify();
    }

    /**
     * 通道就绪事件
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        this.ctx = ctx;
    }

    @Override
    public synchronized Object call() throws Exception {
        // 消息发送
        ctx.writeAndFlush(requestMessage);
        // 线程等待
        wait();
        return responseMessage;
    }
}

RpcClientProxy

/**
 * 客户端代理类，创建代理对象
 * 1、封装request请求对象
 * 2、创建RpcClient对象
 * 3、发送消息
 * 4、返回结果
 *
 * @name: RpcClientProxy
 * @author: terwer
 * @date: 2022-03-13 23:45
 **/
public class RpcClientProxy {
    public static Object createProxy(Class serviceClass) {
        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), new Class[]{serviceClass}, new InvocationHandler() {
            @Override
            public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                // 1、封装request请求对象
                RpcRequest rpcRequest = new RpcRequest();
                rpcRequest.setRequestId(UUID.randomUUID().toString());
                rpcRequest.setClassName(method.getDeclaringClass().getName());
                rpcRequest.setMethodName(method.getName());
                rpcRequest.setParameterTypes(method.getParameterTypes());
                rpcRequest.setParameters(args);

                // 2、创建RpcClient对象
                RpcClient rpcClient = new RpcClient("127.0.0.1", 9999);

                try {
                    // 3、发送消息
                    Object responseMessage = rpcClient.send(JSON.toJSONString(rpcRequest));

                    // 4、返回结果
                    RpcResponse response = JSON.parseObject(responseMessage.toString(), RpcResponse.class);
                    if (response.getError() != null) {
                        throw new RuntimeException(response.getError());
                    }
                    Object result = response.getResult();
                    Object object = JSON.parseObject(result.toString(), method.getReturnType());
                    return object;
                } catch (Exception e) {
                    throw e;
                } finally {
                    rpcClient.close();
                }


            }
        });
    }
}

ClientBoosStrap

/**
 * 客户端启动类
 *
 * @name: ClientBootStrap
 * @author: terwer
 * @date: 2022-03-14 00:00
 **/
public class ClientBootStrap {
    public static void main(String[] args) {
        IUSerService userService = (IUSerService) RpcClientProxy.createProxy(IUSerService.class);
        User user = userService.getById(1);
        System.out.println(user);
    }
}

custom-rpc-provider

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>custom-rpc</artifactId>
        <groupId>com.terewrgreen</groupId>
        <version>1.0.0</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>custom-rpc-provider</artifactId>

    <name>custom-rpc-provider</name>
    <url>http://www.terwergreen.com</url>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <maven.compiler.source>1.8</maven.compiler.source>
        <maven.compiler.target>1.8</maven.compiler.target>
    </properties>

    <dependencies>
        <dependency>
            <groupId>com.terewrgreen</groupId>
            <artifactId>custom-rpc-api</artifactId>
            <version>1.0.0</version>
        </dependency>
        <!--Spring相关依赖 -->
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-context</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-autoconfigure</artifactId>
        </dependency>

        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.13.1</version>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <build>
        <pluginManagement>
        </pluginManagement>
    </build>
</project>

UserServiceImpl

/**
 * 用户服务实现类
 *
 * @name: UserServiceImpl
 * @author: terwer
 * @date: 2022-03-09 23:34
 **/
@RpcService
@Service
public class UserServiceImpl implements IUSerService {
    Map<Object, User> userMap = new HashMap<>();

    @Override
    public User getById(int id) {
        User user = new User();
        user.setId(1);
        user.setName("唐有炜");
        userMap.put(user.getId(), user);

        User user2 = new User();
        user2.setId(2);
        user2.setName("张三");
        userMap.put(user2.getId(), user2);

        return userMap.get(id);
    }
}

RpcServer

/**
 * 对外服务
 *
 * @name: RpcServer
 * @author: terwer
 * @date: 2022-03-09 23:53
 **/
@Service
public class RpcServer implements DisposableBean {
    private NioEventLoopGroup bossGroup;
    private NioEventLoopGroup workerGroup;

    @Autowired
    private RpcServerHandler rpcServerHandler;

    public void startServer(String ip, int port) {
        try {
            // 1、创建线程组
            bossGroup = new NioEventLoopGroup(1);
            workerGroup = new NioEventLoopGroup();

            // 2、创建服务端启动助手
            ServerBootstrap serverBootstrap = new ServerBootstrap();

            // 3、设置参数
            serverBootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel channel) throws Exception {
                            ChannelPipeline pipeline = channel.pipeline();
                            // 添加String的编解码器
                            pipeline.addLast(new StringDecoder());
                            pipeline.addLast(new StringEncoder());
                            // 业务处理类
                            pipeline.addLast(rpcServerHandler);
                        }
                    });

            // 4、绑定端口
            ChannelFuture sync = serverBootstrap.bind(ip, port).sync();
            System.out.println("===========服务端启动成功=============");
            sync.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            if (bossGroup != null) {
                bossGroup.shutdownGracefully();
                System.out.println("finally bossGroup成功关闭");
            }
            if (workerGroup != null) {
                workerGroup.shutdownGracefully();
                System.out.println("finally workerGroup成功关闭");
            }
        }
    }

    @Override
    public void destroy() throws Exception {
        if (bossGroup != null) {
            bossGroup.shutdownGracefully();
            System.out.println("destroy bossGroup成功关闭");
        }
        if (workerGroup != null) {
            workerGroup.shutdownGracefully();
            System.out.println("destroy workerGroup成功关闭");
        }
    }
}

RpcServerHandler

/**
 * 服务端处理类
 * <p>
 * 1、将标有@RpcService注解的类进行缓存
 * 2、接收客户端请求
 * 3、根据传过来的beanName在缓存中查找对应的bean
 * 4、解析请求中的方法名、参数类型、参数信息
 *
 * @name: RpcServerHandler
 * @author: terwer
 * @date: 2022-03-10 00:22
 **/
@Component
@ChannelHandler.Sharable
public class RpcServerHandler extends SimpleChannelInboundHandler<String> implements ApplicationContextAware {
    private static final Map SERVICE_INSTANCE_MAP = new ConcurrentHashMap();

    /**
     * 1、将标有@RpcService注解的类进行缓存
     *
     * @param applicationContext
     * @throws BeansException
     */
    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        Map<String, Object> serviceMap = applicationContext.getBeansWithAnnotation(RpcService.class);
        if (serviceMap != null && serviceMap.size() > 0) {
            Set<Map.Entry<String, Object>> entries = serviceMap.entrySet();
            for (Map.Entry<String, Object> entry : entries) {
                Object serviceBean = entry.getValue();
                if (serviceBean.getClass().getInterfaces().length == 0) {
                    throw new RuntimeException("服务必须实现接口");
                }

                // 默认取第一个接口作为名称
                SERVICE_INSTANCE_MAP.put(serviceBean.getClass().getInterfaces()[0].getName(), serviceBean);
            }

        }
    }

    /**
     * 通道读取就绪事件
     *
     * @param channelHandlerContext
     * @param s
     * @throws Exception
     */
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
        // 接收客户端请求，转换成RpcReuest
        RpcRequest rpcRequest = JSON.parseObject(msg, RpcRequest.class);
        RpcResponse rpcResponse = new RpcResponse();
        rpcResponse.setRequestId(rpcRequest.getRequestId());

        try {
            Object result = handler(rpcRequest);
            rpcResponse.setResult(result);
        } catch (Exception e) {
            rpcResponse.setError(e.getMessage());
            e.printStackTrace();
        }

        ctx.writeAndFlush(JSON.toJSONString(rpcResponse));
    }

    /**
     * 业务逻辑处理方法
     *
     * @param rpcRequest
     * @return
     */
    private Object handler(RpcRequest rpcRequest) throws InvocationTargetException {
        // 根据传过来的beanName在缓存中查找对应的bean
        Object serviceBean = SERVICE_INSTANCE_MAP.get(rpcRequest.getClassName());
        if(null == serviceBean){
            throw new RuntimeException("根据beanName找不到服务"+rpcRequest.getClassName());
        }

        // 解析请求中的方法名、参数类型、参数信息
        Class<?> beanClass = serviceBean.getClass();
        String methodName = rpcRequest.getMethodName();
        Class<?>[] parameterTypes = rpcRequest.getParameterTypes();
        Object[] parameters = rpcRequest.getParameters();

        // 反射调用
        FastClass fastClass = FastClass.create(beanClass);
        FastMethod fastMethod = fastClass.getMethod(methodName, parameterTypes);
        Object result = fastMethod.invoke(serviceBean, parameters);

        return result;
    }
}

ServerBootdtrapApplication

/**
 * 启动类
 *
 * @name: ServerBootstrapApplication
 * @author: terwer
 * @date: 2022-03-09 23:46
 **/
@SpringBootApplication
public class ServerBootstrapApplication implements CommandLineRunner {
    @Autowired
    private RpcServer rpcServer;

    public static void main(String[] args) {
        SpringApplication.run(ServerBootstrapApplication.class, args);
    }

    @Override
    public void run(String... args) throws Exception {
        new Thread(new Runnable() {
            @Override
            public void run() {
                rpcServer.startServer("127.0.0.1", 9999);
            }
        }).start();
    }
}

运行效果

<img src="https://raw.githubusercontent.com/terwer/upload/main/img/20220314005322.png" alt="image-20220314001934218" style="zoom:50%;" />

<img src="https://raw.githubusercontent.com/terwer/upload/main/img/20220314005333.png" alt="image-20220314002004686" style="zoom:50%;" />

错误解决

com.terewrgreen.rpc.provider.handler.RpcServerHandler is not a @Sharable handler, so can't be added or removed multiple times.

加上 @ChannelHandler.Sharable 注解即可。