当前位置：移动技术网 > IT编程>开发语言>Java > 荐我的架构梦：（二十）基于Netty手写RPC框架

荐我的架构梦：（二十）基于Netty手写RPC框架

2020年07月15日 | 移动技术网IT编程 | 我要评论

基于Netty手写RPC框架

一、前言
二、需求与步骤
三、代码实现
四、结果测试
五、代码仓库

一、前言

RPC又称远程过程调用，我们所知的远程调用分为两种，现在在服务间通信的方式也基本以这两种为主：

是基于HTTP的restful形式的广义远程调用，以spring could的feign和restTemplate为代表，采用的协议是HTTP 的7层调用协议，并且协议的参数和响应序列化基本以JSON格式和XML格式为主。
是基于TCP的狭义的RPC远程调用，以阿里的Dubbo为代表，主要通过netty来实现4层网络协议，NIO来异步传输，序列化也可以是JSON或者hessian2以及java自带的序列化等，可以配置。

接下来我们主要以第二种的RPC远程调用来自己实现

二、需求与步骤

模仿 dubbo，消费者和提供者约定接口和协议，消费者远程调用提供者，提供者返回一个字符串，消费者打印提供者返回的数据。底层网络通信使用 Netty。

1、创建一个公共的接口项目以及创建接口及方法，用于消费者和提供者之间的约定。
2、创建一个提供者，该类需要监听消费者的请求，并按照约定返回数据。
3、创建一个消费者，该类需要透明的调用自己不存在的方法，内部需要使用 Netty 请求提供者返回数据。

三、代码实现

1、maven聚合工程

rpc-netty父pom依赖：

<?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">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.3.1.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.riemann</groupId>
    <artifactId>rpc-netty</artifactId>
    <packaging>pom</packaging>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <java.version>1.8</java.version>
    </properties>

    <modules>
        <module>rpc-common</module>
        <module>rpc-provider</module>
        <module>rpc-consumer</module>
    </modules>

    <dependencies>
        <dependency>
            <groupId>io.netty</groupId>
            <artifactId>netty-all</artifactId>
            <version>4.1.16.Final</version>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.12</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.41</version>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
            <exclusions>
                <exclusion>
                    <groupId>org.junit.vintage</groupId>
                    <artifactId>junit-vintage-engine</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>

</project>

rpc-common pom依赖

<?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>rpc-netty</artifactId>
        <groupId>com.riemann</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>rpc-common</artifactId>
    
</project>

rpc-provider pom依赖

<?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>rpc-netty</artifactId>
        <groupId>com.riemann</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>rpc-provider</artifactId>

    <dependencies>
        <dependency>
            <groupId>com.riemann</groupId>
            <artifactId>rpc-common</artifactId>
            <version>1.0-SNAPSHOT</version>
        </dependency>
    </dependencies>

</project>

rpc-consumer pom依赖

<?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>rpc-netty</artifactId>
        <groupId>com.riemann</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>rpc-consumer</artifactId>

    <dependencies>
        <dependency>
            <groupId>com.riemann</groupId>
            <artifactId>rpc-common</artifactId>
            <version>1.0-SNAPSHOT</version>
        </dependency>
    </dependencies>

</project>

2、rpc-common

提供者及消费者工程都需依赖公共模块，这样提供者来实现接口并且提供网络调用，消费者直接通过接口来进行TCP通信及一定的协议定制获取提供者的实现返回值。

public interface IUserService {

    String sayHello(String msg);

}

根据RpcRequest实体作为通信协议

@Data
public class RpcRequest {

    /**
     * 请求对象的ID
     */
    private String requestId;

    /**
     * 类名
     */
    private String className;

    /**
     * 方法名
     */
    private String methodName;

    /**
     * 参数类型
     */
    private Class<?>[] parameterTypes;

    /**
     * 入参
     */
    private Object[] parameters;

}

采用JSON的方式，定义JSONSerializer的实现类。

public class JSONSerializer implements Serializer {

    public byte[] serialize(Object object) throws IOException {
        return JSON.toJSONBytes(object);
    }

    public <T> T deserialize(Class<T> clazz, byte[] bytes) throws IOException {
        return JSON.parseObject(bytes, clazz);
    }

}

3、rpc-provider

首先是接口的实现，这一点和普通接口实现是一样的

@Service
public class UserServiceImpl implements IUserService {

    // 将来客户端要远程调用的方法
    public String sayHello(String msg) {
        System.out.println("hello " + msg);
        return "success";
    }

    // 创建一个方法启动服务器
    public static void startServer(String ip, int port) throws InterruptedException {
        // 1.创建两个线程池对象
        NioEventLoopGroup bossGroup = new NioEventLoopGroup();
        NioEventLoopGroup workGroup = new NioEventLoopGroup();

        // 2.创建服务端的启动引导对象
        ServerBootstrap serverBootstrap = new ServerBootstrap();

        // 3.配置启动引导对象
        serverBootstrap.group(bossGroup, workGroup)
                // 设置通道为NIO
                .channel(NioServerSocketChannel.class)
                // 创建监听channel
                .childHandler(new ChannelInitializer<NioSocketChannel>() {
                    protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {
                        // 获取管道对象
                        ChannelPipeline pipeline = nioSocketChannel.pipeline();
                        // 给管道对象pipLine 设置编码
                        // pipeline.addLast(new StringEncoder());
                        // pipeline.addLast(new StringDecoder());

                        pipeline.addLast(new StringEncoder());
                        pipeline.addLast(new RpcDecoder(RpcRequest.class, new JSONSerializer()));

                        // 把我们自定义的一个ChannelHandler添加到通道中
                        pipeline.addLast(new UserServiceHandler());
                    }
                });

        // 4.绑定端口
        serverBootstrap.bind(ip, port).sync();

        System.out.println("rpc-provider已启动...");
    }

}

在实现中加入了netty的服务器启动程序，上面的代码中添加了String类型的编解码 handler，添加了一个自定义 handler。

自定义 handler 逻辑如下:

/**
 * 自定义的业务处理器
 */
@Component
public class UserServiceHandler extends ChannelInboundHandlerAdapter {

    // 当客户端读取数据时，该方法会被调用
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("接收到客户端信息：" + JSON.toJSON(msg).toString());
        // 将读到的msg对象，强转成RpcRequest对象
        RpcRequest rpcRequest = (RpcRequest) msg;
        // 加载class文件
        Class<?> clazz = Class.forName(rpcRequest.getClassName());
        // 通过class获取服务器spring容器中service类的实例化bean对象
        Object serviceBean = SpringContextUtil.getBean(clazz);
        Method method = clazz.getMethod(rpcRequest.getMethodName(), rpcRequest.getParameterTypes());
        method.invoke(serviceBean, rpcRequest.getParameters());

        //服务器写入数据，将结果返回给客户端
        ctx.writeAndFlush("success");

        // 注意：客户端将来发送请求的时候会传递一个参数：UserService#sayHello#riemann

        // 1.判断当前的请求是否符合规则
        /*if (msg.toString().startsWith("UserService")) {
            // 2.如何符合规则，调用实现类获取到一个result
            UserServiceImpl userService = new UserServiceImpl();
            String result = userService.sayHello(msg.toString().substring(msg.toString().lastIndexOf("#") + 1));
            // 3.将调用实现类的方法获得的结果写到客户端
            ctx.writeAndFlush(result);
        }*/
    }

}

还需要一个启动类:

@SpringBootApplication
public class ServerBoot {

    public static void main(String[] args) throws InterruptedException {
        SpringApplication.run(ServerBoot.class, args);
        // 启动服务器
        UserServiceImpl.startServer("127.0.0.1", 8999);
    }

}

4、rpc-consumer

消费者有一个需要注意的地方，就是调用需要透明，也就是说，框架使用者不用关心底层的网络实现。这里我们可以使用 JDK 的动态代理来实现这个目的。

思路：客户端调用代理方法，返回一个实现了 HelloService 接口的代理对象，调用代理对象的方法，返回结果。

我们需要在代理中做手脚，当调用代理方法的时候，我们需要初始化 Netty 客户端，还需要向服务端请求数据，并返回数据。

首先创建代理相关的类：

/**
 * 消费者
 */
public class RPCConsumer {

    // 1.创建一个线程池对象  -- 它要处理我们自定义事件
    private static ExecutorService executorService =
            Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

    // 2.声明一个自定义事件处理器  UserClientHandler
    private static UserClientHandler userClientHandler;

    // 3.编写方法，初始化客户端 （创建连接池、bootStrap、设置bootStrap、连接服务器）
    public static void initClient() throws InterruptedException {
        // 1) 初始化 userClientHandler
        userClientHandler = new UserClientHandler();
        // 2) 创建连接池对象
        EventLoopGroup group = new NioEventLoopGroup();
        // 3) 创建客户端的引导对象
        Bootstrap bootstrap = new Bootstrap();
        // 4) 配置启动引导对象
        bootstrap.group(group)
                // 设置通道为NIO
                .channel(NioSocketChannel.class)
                // 设置请求协议为TCP
                .option(ChannelOption.TCP_NODELAY, true)
                // 监听channel 并初始化
                .handler(new ChannelInitializer<SocketChannel>() {
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        // 获取 ChannelPipeline
                        ChannelPipeline pipeline = socketChannel.pipeline();
                        // 设置编码
                        // pipeline.addLast(new StringEncoder());
                        // pipeline.addLast(new StringDecoder());

                        pipeline.addLast(new RpcEncoder(RpcRequest.class, new JSONSerializer()));
                        pipeline.addLast(new StringDecoder());
                        // 添加自定义事件处理器
                        pipeline.addLast(userClientHandler);
                    }
                });
        // 5) 连接服务端
        bootstrap.connect("127.0.0.1", 8999).sync();
        System.out.println("rpc-consumer已启动...");
    }

    // 4.编写一个方法，使用JDK的动态代理创建对象
    /**
     *
     * @param serviceClass   接口类型，根据哪个接口生成子类代理对象
     * @param providerParam  "UserService#sayHello#riemann"
     * @return
     */
    public static Object createProxy(Class<?> serviceClass, final String providerParam) {
        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(),
                new Class[]{serviceClass}, new InvocationHandler() {
                    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                        // 1.初始化客户端client
                        if (userClientHandler == null) {
                            initClient();
                        }

                        String requestId = UUID.randomUUID().toString();
                        String className = method.getDeclaringClass().getName();
                        String methodName = method.getName();
                        Class<?>[] parameterTypes = method.getParameterTypes();

                        RpcRequest rpcRequest = new RpcRequest();
                        rpcRequest.setRequestId(requestId);
                        rpcRequest.setClassName(className);
                        rpcRequest.setMethodName(methodName);
                        rpcRequest.setParameterTypes(parameterTypes);
                        rpcRequest.setParameters(args);

                        // 2.给 userClientHandler 设置param参数
                        // userClientHandler.setParam(providerParam + args[0]);
                        userClientHandler.setParam(rpcRequest);

                        // 3.使用线程池，开启一个线程处理call()写操作，并返回结果。
                        Object result = executorService.submit(userClientHandler).get();

                        // 4.return 结果
                        return result;
                    }
                });
    }

}

该类有 2 个方法，创建代理和初始化客户端。

创建代理逻辑:使用 JDK 的动态代理技术，代理对象中的 invoke 方法实现如下: 如果 client 没有初始化，则初始化 client，这个 client 既是 handler ，也是一个 Callback。将参数设置进 client ，使用线程池调用 client 的 call 方法并阻塞等待数据返回。

初始化客户端逻辑: 创建一个 Netty 的客户端，并连接提供者，并设置一个自定义 handler，和一些 String 类型的序列化方式。

UserClientHandler 的实现:

/**
 * 自定义事件处理器
 */
public class UserClientHandler extends ChannelInboundHandlerAdapter implements Callable {

    // 1.定义成员变量
    private ChannelHandlerContext context; // 事件处理器上下文对象（存储handler写信息，写操作）
    private String result; // 记录服务器返回的数据
    private Object param; // 记录将要返回给服务器的数据

    // 2.实现 channelActive 客户端和服务器连接时，该方法就自动执行。
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("服务器已连接...");
        // 初始化 ChannelHandlerContext
        this.context = ctx;
    }


    // 3.实现 channelRead 当我们读到服务器数据，该方法自动执行。
    @Override
    public synchronized void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        // 将读到的服务器的数据msg,设置为成员变量的值
        result = msg.toString();
        notify();
    }

    // 4.将客户端的数据写到服务器
    public synchronized Object call() throws Exception {
        // context 给服务器写数据
        context.writeAndFlush(param);
        wait();
        return result;
    }

    // 5.设置参数的方法
    public void setParam(Object param) {
        this.param = param;
    }
}

该类缓存了 ChannelHandlerContext，用于下次使用，有两个属性：返回结果和请求参数。

当成功连接后，缓存 ChannelHandlerContext，当调用 call 方法的时候，将请求参数发送到服务端，等待。当服务端收到并返回数据后，调用 channelRead 方法，将返回值赋值个 result，并唤醒等待在 call 方法上的线程。此时，代理对象返回数据。

再看看消费者调用方式，一般的TCP的RPC只需要这样调用即可，无需关心具体的协议和通信方式:

public class ConsumeBoot {

    // 参数定义
    private static final String PROVIDE_NAME = "UserService#sayHello#";

    public static void main(String[] args) throws InterruptedException {
        // 1.创建代理对象
        IUserService service = (IUserService) RPCConsumer.createProxy(IUserService.class, PROVIDE_NAME);

        // 2.循环给服务器写数据
        while (true) {
            String result = service.sayHello("riemann");
            System.out.println(result);
            Thread.sleep(2000);
        }
    }

}

四、结果测试

调用者首先创建了一个代理对象，然后每隔一秒钟调用代理的 sayHello 方法，并打印服务端返回的结果。

服务提供者：
在这里插入图片描述
服务消费者：

可以看到，消费者无需通过jar包的形式引入具体的实现项目，而是通过远程TCP通信的形式，以一定的协议和代理通过接口直接调用了方法，实现远程service间的调用，是分布式服务的基础。

五、代码仓库

https://github.com/riemannChow/perseverance/tree/master/handwriting-framework/rpc-netty

本文地址：https://blog.csdn.net/riemann_/article/details/107350656

您可能感兴趣的文章:

如对本文有疑问，点击进行留言回复！！

关于msyql事务隔离你要知道

什么是事务？事务是数据库管理系统执行过程中的一个逻辑单位，由一个有限的数据库操作序列构成。数据库事务通常包含了一个序列的对数据库的读/写操作。包含有以下两个目的... [阅读全文]
JavaFX实现UI美观效果代码实例

相对于swing来说，javafx在ui上改善了很多，不仅可以通过fxml来排版布局界面，同时也可以通过css样式表来美化ui。其实在开发javafx应用的时候... [阅读全文]
使用java实现网络爬虫

接着上面一篇，这一篇目的就是在于网络爬虫的实现，对数据的获取，以便分析。----->爬虫实现原理网络爬虫基本技术处理网络爬虫是数据采集的一种方法，实际项目... [阅读全文]
JavaFX桌面应用未响应问题解决方案

日常使用软件的过程中，偶尔会遇到软件突然卡住，再点击几次就变成“未响应”的情况。在javafx应用中同样也会出现这种情况，在开发过程中应该尽量避免这种情况的出现... [阅读全文]
java调用回调机制详解

调用和回调机制在一个应用系统中, 无论使用何种语言开发, 必然存在模块之间的调用, 调用的方式分为几种:1.同步调用同步调用是最基本并且最简单的一种调用方式, ... [阅读全文]
Springboot项目因为kackson版本问题启动报错解决方案

问题现象org.springframework.context.applicationcontextexception: unable to start emb... [阅读全文]
idea中database不显示问题的解决

【问题】一般情况下，database会显示在idea的最右边，就像这个样子：一无所有。。。【理想界面】：【解决方法】方法一：1）点击view 2)点击tool ... [阅读全文]
Java多线程下的其他组件之CyclicBarrier、Callable、Future和FutureTask详解

cyclicbarrier 接着讲多线程下的其他组件，第一个要讲的就是cyclicbarrier。cyclicbarrier从字面理解是指循环屏障，它可以协同多... [阅读全文]
IDEA POJO开发神器之Groovy的使用详解

暂时只对 mysql进行了测试项目使用 lombok mybatis-plus一：使用步骤首先在项目右侧找到 database 如图没有请参考 2.点开之后进... [阅读全文]
idea+git合并分支解决冲突及详解步骤

git分支详解参考：分支管理组成1.1、master主干在版本管理中，代码库应该仅有一个主干。此主干是和当前生产保持一致的，是可用的、稳定的可直接发布的版本，不... [阅读全文]