手动实现简易版RPC(下)

手动实现简易版RPC(下)

前言

什么是RPC?它的原理是什么?它有什么特点?如果让你实现一个RPC框架,你会如何是实现?带着这些问题,开始今天的学习。

接上一篇博客 手动实现简易版RPC(上) 我们得到了最简易RPC框架的架构图与运行图,本文主要介绍简易版RPC 的简易实现。


架构

最简易的 RPC 框架架构图:

RPC框架

整个简易RPC的调用过程图

RPC的调用过程图

下面我们白手起家,从0实现简易版的RPC框架

项目准备

1、创建项目

优先创建一个名为ape-rpc的空项目,然后使用idea 依次创建几个maven模块

在这里插入图片描述

整个项目包结构也如上图所示,分别介绍一下上述的包结构

  • example-common 整个项目中所用到的公共类,例如一些实体类,公共接口等
  • example-producer 示例服务生产者代码包
  • example-consumer 示例服务消费者代码包
  • ape-rpc-easy 简易RPC框架实现代码包

在示例项目中,我们将以一个最简单的猫咪服务为例,演示整个服务调用过程。下面我们依次实现上述的几个模块。

2、common模块代码实现

公共模块需要同时被服务消费者服务提供者引入,主要是编写和服务相关的接口和数据实体

common模块结构如下图

在这里插入图片描述

2.1 引入lombok 简化开发
  	<dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.24</version>
        </dependency>
2.2 编写🐱实体
package com.jerry.common.model;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.ToString;

import java.io.Serializable;

/**
 * @version 1.0
 * @Author jerryLau
 * @Date 2024/4/9 15:33
 * @注释 实体 🐱
 */
@Data
@AllArgsConstructor
@NoArgsConstructor
@ToString
public class Cat implements Serializable {

    private int id;
    /***
     * 名字
     */
    private String name;

    /***
     * 颜色
     */
    private String color;
}

注意,对象需要实现序列化接口,为后续网络传输序列化提供支持

2.3编写猫咪服务接口

编写猫咪服务接口CatService,提供两个获取猫咪的方法

package com.jerry.common.service;

import com.jerry.common.model.Cat;

/**
 * @version 1.0
 * @Author jerryLau
 * @Date 2024/4/9 15:38
 * @注释 🐱服务接口
 */

public interface CatService {
    /***
     * 获取猫咪信息
     * @param cat
     * @return
     */
    Cat getCat(Cat cat);


    /***
     * 按照id获取猫咪信息
     * @param id
     * @return
     */
    Cat getCatById(int id);

    //....other
}

3、服务生产者(producer)

作为服务生产者,是真正实现服务接口的模块,需要调用简单RPC模块

3.1添加pom依赖
 <dependencies>
     	<!--common-->
        <dependency>
            <groupId>com.jerry</groupId>
            <artifactId>example-common</artifactId>
            <version>1.0-SNAPSHOT</version>
        </dependency>
		<!--rpc-easy-->
        <dependency>
            <groupId>com.jerry</groupId>
            <artifactId>ape-rpc-easy</artifactId>
            <version>1.0-SNAPSHOT</version>
        </dependency>
		<!--hutool工具包-->
        <dependency>
            <groupId>cn.hutool</groupId>
            <artifactId>hutool-all</artifactId>
            <version>5.8.16</version>
        </dependency>
		<!--lombok-->
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.24</version>
        </dependency>
    </dependencies>
3.2 编写服务实现类

编写代码实现类,实现公共模块中定义的🐱服务接口

功能是打印🐱的信息

package com.jerry.producer.serviceImpl;

import com.jerry.common.model.Cat;
import com.jerry.common.service.CatService;

/**
 * @version 1.0
 * @Author jerryLau
 * @Date 2024/4/9 15:45
 * @注释 数据提供者实现
 */
public class CatServiceImpl implements CatService {
    @Override
    public Cat getCat(Cat cat) {
        System.out.println("调用到了服务提供者:" + cat.toString());
        return cat;
    }

    @Override
    public Cat getCatById(int id) {
        Cat tom = new Cat(id, "TOM", "#FFFFFF");
        System.out.println("调用到了服务提供者:" + tom.toString());
        return tom;
    }
}

3.3实现生产者主类

编写服务提供者启动类EasyProducer ,之后会在该类的 main 方法中编写提供服务的代码。

package com.jerry.producer;

import com.jerry.aperpc.localregcenter.LocalRegCenter;
import com.jerry.aperpc.server.VertxHttpServer;
import com.jerry.common.service.CatService;
import com.jerry.producer.serviceImpl.CatServiceImpl;

/**
 * @version 1.0
 * @Author jerryLau
 * @Date 2024/4/9 15:47
 * @注释 简单的生产者
 */
public class EasyProducer {
    public static void main(String[] args) {
        //提供服务....
     
    }
}

完成后代码生产者的结构大概如下图所示

在这里插入图片描述

4、服务消费者(Consumer)

服务消费者需要调用简单rpc实现模块

4.1pom依赖配置
 <dependencies>
        <dependency>
            <groupId>com.jerry</groupId>
            <artifactId>example-common</artifactId>
            <version>1.0-SNAPSHOT</version>
        </dependency>

        <dependency>
            <groupId>com.jerry</groupId>
            <artifactId>ape-rpc-easy</artifactId>
            <version>1.0-SNAPSHOT</version>
        </dependency>

        <dependency>
            <groupId>cn.hutool</groupId>
            <artifactId>hutool-all</artifactId>
            <version>5.8.16</version>
        </dependency>

        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.24</version>
        </dependency>
    </dependencies>
4.2创建服务消费者启动类

创建服务消费者启动类EasyConsumer,编写调用接口的代码

package com.jerry.consumer;

import com.jerry.aperpc.proxy.ServiceProxyFactory;
import com.jerry.common.model.Cat;
import com.jerry.common.service.CatService;

/**
 * @version 1.0
 * @Author jerryLau
 * @Date 2024/4/9 17:14
 * @注释
 */
public class EasyConsumer {
    public static void main(String[] args) {
        //todo: 需要获取 CatService 的实现类对象
        CatService catService = null;

        //调用
        Cat newCat = catService.getCatById(1);
        if (newCat != null) {
            System.out.println("消费者获取到的 cat :" + newCat.toString());
        } else {
            System.out.println("no cat");
        }

    }

}

值得注意的是:现在是无法获取到 CatService实例的,所以预留为 null。我们之后的目标是,能够通过 RPC 框架快速得到一个支持远程调用服务提供者的代理对象,像调用本地方法一样调用 CatService的方法。

5、简单rpc业务实现

5.1 web服务器

我们要先让服务提供者提供可远程访问的服务。那么,就需要一个 web 服务器,能够接受处理请求、并返回响应。
web 服务器的选择有很多,比如 Spring Boot 内嵌的 Tomcat、NIO 框架 Netty 和 Vert.x等等

此处我们使用高性能的 NIO 框架 Vert.x 来作为 RPC 框架的 web 服务器。

想了解更多,请参考 Vert.x官方文档🌐 || Vert.x官方文档中文版🌐

5.1.1 rpc-easy引入pom依赖
<dependencies>
        <!--        高性能的 NIO 框架 Vert.x 来作为 RPC 框架的 web 服务器。-->
        <dependency>
            <groupId>io.vertx</groupId>
            <artifactId>vertx-core</artifactId>
            <version>4.5.1</version>
        </dependency>

        <dependency>
            <groupId>cn.hutool</groupId>
            <artifactId>hutool-all</artifactId>
            <version>5.8.16</version>
        </dependency>

        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.24</version>
        </dependency>
    </dependencies>
5.1.2编写httpServer做web服务器接口

编写一个 web 服务器的接口 HttpServer,定义统一的启动服务器方法,便于后续的扩展,比如实现多种不同的
web 服务器。

package com.jerry.aperpc.server;

/**
 * @version 1.0
 * @Author jerryLau
 * @Date 2024/4/9 16:03
 * @注释
 */
public interface HttpServer {

    /***
     * 启动器
     * @param port
     */
    void exec(int port);
}
5.1.3基于vert.x实现请求处理

编写基于 Vert.x 实现的 web 服务器 VertxHttpServer,能够监听指定端口并处理请求。

package com.jerry.aperpc.server;

import io.vertx.core.Vertx;

/**
 * @version 1.0
 * @Author jerryLau
 * @Date 2024/4/9 16:06
 * @注释
 */
public class VertxHttpServer implements HttpServer {
    @Override
    public void exec(int port) {
        //1-创建Vertx 实例
        Vertx vertx = Vertx.vertx();

        //2-创建http服务器
        io.vertx.core.http.HttpServer httpServer = vertx.createHttpServer();

        //3-监听端口并处理请求
       httpServer.requestHandler(httpServerRequest -> {
           //处理里HTTP 请求
           System.out.println("Received request :"
                  + httpServerRequest.method() + " " +
                   httpServerRequest.uri());

           //发送http响应
           httpServerRequest.response()
                   .putHeader("content-type", "text/plain")
                  .end("Hello from Vert.x HTTP server!");
       });
       

        //4-启动 HTTP 服务器并监听指定端口
        httpServer.listen(port, result -> {
            if (result.succeeded()) {
                System.out.println("Server is now listening on port:" + port);
            } else
                System.err.println("Failed to start server:" + result.cause());

        });
    }
}
5.1.4验证web服务器

验证 web 服务器能否启动成功并接受请求。

修改服务提供者(example-producer)模块的 EasyProducer类,编写启动 web 服务的代码,如下:

public class EasyProducer {
    public static void main(String[] args) {
      
        //提供服务
        VertxHttpServer vertxHttpServer = new VertxHttpServer();
        vertxHttpServer.exec(8080);
    }
}

浏览器访问localhost:8080,查看能否正常访问并看到输出的文字。

如果能够正常访问,浏览器窗口以及控制台输出如下图所示

在这里插入图片描述

在这里插入图片描述

5.2 本地服务注册器

我们现在做的简易 RPC 框架主要是跑通流程,所以暂时先不用第三方注册中心,直接把服务注册到服务提供者本地即可。

在 rpc-easy 模块中新建文件夹 localregcenter ,创建本地服务注册器 LocalRegCenter,使用线程安全的 ConcurrentHashMap 存储服务注册信息,key 为服务名称、value 为服务的实现类。之后就可以根据要调用的服务名称获取到对应的实现类,然后通过反射进行方法调用了。

具体代码如下:

package com.jerry.aperpc.localregcenter;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @version 1.0
 * @Author jerryLau
 * @Date 2024/4/9 16:23
 * @注释 本地服务注册器
 */
public class LocalRegCenter {
    /***
     * 本地注册中心存储列表
     */
    private static Map<String, Class<?>> map = new ConcurrentHashMap<String, Class<?>>();


    /***
     * 注册
     * @param serviceName
     * @param impl
     */
    public static void add(String serviceName, Class<?> impl) {
        map.put(serviceName, impl);
    }

    /***
     * 获取
     * @param serviceName
     * @return
     */
    public static Class<?> get(String serviceName) {
        return map.get(serviceName);
    }


    /***
     * 移除
     * @param serviceName
     */
    public static void remove(String serviceName) {
        map.remove(serviceName);
    }

}

注意:本地服务注册器和注册中心的作用是有区别的。

注册中心的作用侧重于管理注册的服务、提供服务信息给消费者;

而本地服务注册器的作用是根据服务名获取到对应的实现类,是完成调用必不可少的模块。

当服务提供者(example-producer)启动时,需要将自己注册服务到注册器中,修改 EasyProducer代码如下

package com.jerry.producer;

import com.jerry.aperpc.localregcenter.LocalRegCenter;
import com.jerry.aperpc.server.VertxHttpServer;
import com.jerry.common.service.CatService;
import com.jerry.producer.serviceImpl.CatServiceImpl;

/**
 * @version 1.0
 * @Author jerryLau
 * @Date 2024/4/9 15:47
 * @注释 简单的生产者
 */
public class EasyProducer {
    public static void main(String[] args) {
        //注册服务
        LocalRegCenter.add(CatService.class.getName(), CatServiceImpl.class);

        //提供服务
        VertxHttpServer vertxHttpServer = new VertxHttpServer();
        vertxHttpServer.exec(8080);
    }
}
5.3 序列化器

服务在本地注册后,我们就可以根据请求信息取出实现类并通过反射技术调用实现类的方法了。

在编写处理请求的逻辑前,我们要先实现序列化器模块。

5.3.1 什么是序列化

无论是请求或响应,都会涉及参数的传输。而 Java对象是存活在JVM 虚拟机中的,如果想在其他位置存储并访问、或者在网络中进行传输,就需要进行序列化和反席列化。

简单理解:

**序列化:**将数据结构或对象转换成二进制字节流的过程

**反序列化:**将在序列化过程中所生成的二进制字节流转换成数据结构或者对象的过程

5.3.2 序列化处理要素

序列化的处理要素

  • **解析效率:**序列化协议应该首要考虑的因素,像xml/json解析起来比较耗时,需要解析doom树,二进制自定义协议解析起来效率要快很多。
  • **压缩率:**同样一个对象,xml/json传输起来有大量的标签冗余信息,信息有效性低,二进制自定义协议占用的空间相对来说会小很多。
  • **扩展性与兼容性:**是否能够利于信息的扩展,并且增加字段后旧版客户端是否需要强制升级,这都是需要考虑的问题,在自定义二进制协议时候,要做好充分考虑设计。
  • **可读性与可调试性:**xml/json的可读性会比二进制协议好很多,并且通过网络抓包是可以直接读取,二进制则需要反序列化才能查看其内容
  • **跨语言:**有些序列化协议是与开发语言紧密相关的,例如dubbo的Hessian序列化协议就只能支持Java的RPC调用。
  • **通用性:**xml/json非常通用,都有很好的第三方解析库,各个语言解析起来都十分方便,二进制数据的处理方面也有Protobu和和Hessian等插件,在做设计的时候尽量做到较好的通用性。
5.3.3 序列化器
  1. JDK原生序列化,通过实现Serializable接口。通过ObjectOutPutSream和ObjectlnputStream对象进行序列化及反序列化.
  2. JSON序列化。一般在HTTP协议的RPC框架通信中,会选择JSON方式。JSON具有较好的扩展性、可读性和通用性。但JSON序列化占用空间开销较大,没有JAVA的强类型区分,需要通过反射解决,解析效率和压缩率都较差。如果对并发和性能要求较高,或者是传输数据量较大的场景,不建议采用JSON序列化方式。
  3. Hessian2序列化。Hessian 是一个动态类型,二进制序列化,并且支持跨语言特性的序列化框架。Hessian 性能上要比JDK、JSON 序列化高效很多,并且生成的字节数也更小。有非常好的兼容性和稳定性,所以 Hessian 更加适合作为 RPC 框架远程通信的序列化协议。
  4. kryo序列化。高性能的序列化/反序列化工具,由于其变长存储特性并使用了字节码生成机制,拥有较高的运行速度和较小的字节码体积
  5. Protobuf序列化。Protobuf 出自于 Google,性能还比较优秀,也支持多种语言,同时还是跨平台的。就是在使用中过于繁琐,因为你需要自己定义 IDL 文件和生成对应的序列化代码。这样虽然不灵活,但是,另一方面导致 protobuf 没有序列化漏洞的风险。
  6. 其他序列化方式

在此我们实现简单的kryo序列化

  • 首先在rpc模块pom中引入kryo系列化依赖
 <dependency>
        <groupId>com.esotericsoftware</groupId>
        <artifactId>kryo</artifactId>
        <version>4.0.1</version>
    </dependency>

由于其底层依赖于ASM技术,与Spring等框架可能会发生ASM依赖的版本冲突(文档中表示这个冲突还挺容易出现)所以提供了另外一个依赖以供解决此问题

<dependency>
        <groupId>com.esotericsoftware</groupId>
        <artifactId>kryo-shaded</artifactId>
        <version>4.0.1</version>
    </dependency>

同时引入异常管理依赖

  <dependency>
            <groupId>com.nimbusds</groupId>
            <artifactId>oauth2-oidc-sdk</artifactId>
            <version>8.36</version>
        </dependency>
  • 然后在rpc模块新建序列化接口Serializer,提供序列化和反序列化两个接口,方便后续拓展
package com.jerry.aperpc.serializer;

import java.io.IOException;

/**
 * @version 1.0
 * @Author jerryLau
 * @Date 2024/4/9 16:33
 * @注释
 */
public interface Serializer {
    /***
     * 序列化
     * @param object
     * @return
     * @param <T>
     * @throws IOException
     */
    <T> byte[] serialize(T object) throws IOException;

    /***
     * 反序列化器
     * @param bytes
     * @param type
     * @return
     * @param <T>
     * @throws IOException
     */
    <T> T deserialize(byte[]bytes, Class<T> type) throws IOException;
}

  • 基于kryo实现KryoSerializer
package com.jerry.aperpc.serializer;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.jerry.aperpc.model.RpcRequest;
import com.jerry.aperpc.model.RpcResponse;
import com.nimbusds.oauth2.sdk.SerializeException;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;

/**
 * @version 1.0
 * @Author jerryLau
 * @Date 2024/4/11 9:08
 * @注释 基于kryo的序列化器
 */
public class KryoSerializer implements Serializer {

    /**
     * Because Kryo is not thread safe. So, use ThreadLocal to store Kryo objects
     */
    private final ThreadLocal<Kryo> kryoThreadLocal = ThreadLocal.withInitial(() -> {
        Kryo kryo = new Kryo();
        kryo.register(RpcResponse.class);
        kryo.register(RpcRequest.class);
        return kryo;
    });

    @Override
    public byte[] serialize(Object obj) {
        try (ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
             Output output = new Output(byteArrayOutputStream)) {
            Kryo kryo = kryoThreadLocal.get();
            // Object->byte:将对象序列化为byte数组
            kryo.writeObject(output, obj);
            kryoThreadLocal.remove();
            return output.toBytes();
        } catch (Exception e) {
            e.printStackTrace();
            throw new SerializeException("Serialization failed");
        }
    }

    @Override
    public <T> T deserialize(byte[] bytes, Class<T> clazz) {
        try (ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(bytes);
             Input input = new Input(byteArrayInputStream)) {
            Kryo kryo = kryoThreadLocal.get();
            // byte->Object:从byte数组中反序列化出对象
            Object o = kryo.readObject(input, clazz);
            kryoThreadLocal.remove();
            return clazz.cast(o);
        } catch (Exception e) {
            e.printStackTrace();
            throw new SerializeException("Deserialization failed");
        }
    }

}

至此序列化器完成

5.4 请求处理器 (生产者处理调用)

请求处理器是RPC框架的核心,其主要功能就是:

根据接收到的请求,并根据请求参数找到对应的服务和方法,通过反射实现调用,最后封装返回结果并响应请求。

5.4.1 rpc模块请求响应实体封装

rpc模块中进行请求以及响应的实体封装

rpc请求体(RpcRequest)

请求类 RpcRequest 的作用是封装调用所需的信息,比如服务名称、方法名称、调用参数的类型列表、参数列表。这些都是 Java 反射机制所需的参数。

package com.jerry.aperpc.model;

import lombok.*;

import java.io.Serializable;

/**
 * @version 1.0
 * @Author jerryLau
 * @Date 2024/4/9 16:43
 * @注释 rpc 请求封装
 */
@Data
@AllArgsConstructor
@NoArgsConstructor
@ToString
@Builder
public class RpcRequest implements Serializable {
    /***
     服务名称
     */
    private String serviceName;
    /***
     *方法名称
     */
    private String methodName;
    /***
     参数类型列表
     */
    private Class<?> [] parameterTypes;
    /***
     *参数列表
     */
    private Object[] args;
}

rpc响应体(RpcResponse)

响应体 RpcResponse 的作用是封装调用方法得到的返回值、以及调用的信息(比如异常情况)等。

package com.jerry.aperpc.model;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.ToString;

import java.io.Serializable;

/**
 * @version 1.0
 * @Author jerryLau
 * @Date 2024/4/9 16:47
 * @注释 rpc 响应封装
 */
@Data
@AllArgsConstructor
@NoArgsConstructor
@ToString
public class RpcResponse implements Serializable {
    /***
     *响应数据
     */
    private Object data;
    /***
     *响应数据类型(预留)
     */
    private Class<?> dataType;
    /***
     *响应信息
     */
    private String message;
    /***
     * 异常信息
     */
    private Exception exception;
}

5.4.2 业务请求处理器

作为业务请求处理器,他有如下几个职责:

1.反序列化请求为对象,并从请求对象中获取参数
2.根据服务名称从本地注册器中获取到对应的服务实现类
3.通过反射机制调用服务实现类中的方法,得到返回结果
4.对返回结果进行封装和序列化,并写入到响应中

完整代码如下

package com.jerry.aperpc.server;

import com.jerry.aperpc.localregcenter.LocalRegCenter;
import com.jerry.aperpc.model.RpcRequest;
import com.jerry.aperpc.model.RpcResponse;
import com.jerry.aperpc.serializer.JDKSerializer;
import com.jerry.aperpc.serializer.KryoSerializer;
import com.jerry.aperpc.serializer.Serializer;
import io.vertx.core.Handler;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.http.HttpServerRequest;
import io.vertx.core.http.HttpServerResponse;

import java.io.IOException;
import java.lang.reflect.Method;

/**
 * @version 1.0
 * @Author jerryLau
 * @Date 2024/4/9 16:49
 * @注释 请求处理器
 * 1.反序列化请求为对象,并从请求对象中获取参数。
 * 2.根据服务名称从本地注册器中获取到对应的服务实现类,
 * 3.通过反射机制调用方法,得到返回结果。
 * 4.对返回结果进行封装和序列化,并写入到响应中,
 */
public class HttpServerHandler implements Handler<HttpServerRequest> {
    @Override
    public void handle(HttpServerRequest request) {
        //1-指定序列化器
        final Serializer serializer = new KryoSerializer();
        //打印请求信息日志
        System.out.println("Received request :" +"request.uri: ["+ request.uri() + "] request.method : ["+request.method()+"]");

        //2-异步进行请求
        request.bodyHandler(body -> {
            byte[] bytes = body.getBytes();
            RpcRequest rpcRequest = null;
            try {
                rpcRequest = serializer.deserialize(bytes, RpcRequest.class);//反序列化请求为对象
            } catch (Exception e) {
                e.printStackTrace();
            }

            // 构造响应结果对象
            RpcResponse rpcResponse = new RpcResponse();
            //如果请求为 nu11,直接返回
            if (rpcRequest == null) {
                rpcResponse.setMessage("rpcRequest is null");
                doResponse(request, rpcResponse, serializer);
                return;
            }


            try {
                // 获取要调用的服务实现类,通过反射调用
                Class<?> implClass = LocalRegCenter.get(rpcRequest.getServiceName());//获取实现类
                Method method = implClass.getMethod(rpcRequest.getMethodName(), rpcRequest.getParameterTypes());//反射获取实现类的方法
                Object result = method.invoke(implClass.newInstance(), rpcRequest.getArgs());//反射执行对应方法
                // 封装返回结果
                rpcResponse.setData(result);
                rpcResponse.setDataType(method.getReturnType());
                rpcResponse.setMessage("ok");
            } catch (Exception e) {
                e.printStackTrace();
                rpcResponse.setMessage(e.getMessage());
                rpcResponse.setException(e);
            }
            // 响应
            doResponse(request, rpcResponse, serializer);

        });

    }

    /**
     * 响应
     *
     * @param request
     * @param rpcResponse
     * @param serializer
     */
    void doResponse(HttpServerRequest request, RpcResponse rpcResponse, Serializer serializer) {
        HttpServerResponse httpServerResponse = request.response()
                .putHeader("content-type", "application/json");
        try {
            // 序列化
            byte[] serialized = serializer.serialize(rpcResponse);
            httpServerResponse.end(Buffer.buffer(serialized));
        } catch (IOException e) {
            e.printStackTrace();
            httpServerResponse.end(Buffer.buffer());
        }
    }
}

简单解释一下上述部分代码:

  1. 获取实现类:
Class<?> implClass = LocalRegCenter.get(rpcRequest.getServiceName());

这行代码从LocalRegCenter(本地服务发现器)中,根据rpcRequest.getServiceName()提供的服务名,获取了对应的实现类的Class对象。这个Class对象代表了实现类的元数据,包括它的方法、字段等信息。

  1. 获取实现类的方法:
Method method = implClass.getMethod(rpcRequest.getMethodName(),rpcRequest.getParameterTypes());

这行代码通过implClass.getMethod(...)方法,根据方法名和参数类型来获取实现类中的某个方法。其中:

  • rpcRequest.getMethodName()提供了方法名。
  • rpcRequest.getParameterTypes()提供了方法的参数类型。

getMethod方法返回的是一个Method对象,代表了这个方法。

  1. 创建实现类的实例:
implClass.newInstance()

通过newInstance方法,创建了实现类的一个新实例。这是Java反射中创建对象的一种方式(注意:从Java 9开始,newInstance方法已被标记为过时,建议使用getDeclaredConstructor().newInstance()来替代)。

  1. 调用实现类的方法:
Object result = method.invoke(implClass.newInstance(), rpcRequest.getArgs());

使用method.invoke(...)方法,调用上面获取到的Method对象所代表的方法。这里做了两件事:

  • 首先,通过implClass.newInstance()创建了一个实现类的新实例。
  • 然后,使用rpcRequest.getArgs()提供的参数来调用这个方法。

invoke方法返回的是方法的返回值,这里被存储在result变量中。

需要注意,不同的 web 服务器对应的请求处理器实现方式也不同,比如 Ver.x 中是通过实现Handler<HitpServerRequest>接口来自定义请求处理器的。并且可以通过 request.bodyHandler 异步处理请求.

5.4.3 给 HttpServer 绑定请求处理器。

修改 VertxHttpServer 的代码,通过server.requestHandler绑定请求处理器

修改后的代码如下

package com.jerry.aperpc.server;

import io.vertx.core.Vertx;

/**
 * @version 1.0
 * @Author jerryLau
 * @Date 2024/4/9 16:06
 * @注释
 */
public class VertxHttpServer implements HttpServer {
    @Override
    public void exec(int port) {
        //1-创建Vertx 实例
        Vertx vertx = Vertx.vertx();

        //2-创建http服务器
        io.vertx.core.http.HttpServer httpServer = vertx.createHttpServer();

        //3-监听端口并处理请求
//        httpServer.requestHandler(httpServerRequest -> {
//            //处理里HTTP 请求
//            System.out.println("Received request :"
//                    + httpServerRequest.method() + " " +
//                    httpServerRequest.uri());
//
//            //发送http响应
//            httpServerRequest.response()
//                    .putHeader("content-type", "text/plain")
//                    .end("Hello from Vert.x HTTP server!");
//        });
        httpServer.requestHandler(new HttpServerHandler());

        //4-启动 HTTP 服务器并监听指定端口
        httpServer.listen(port, result -> {
            if (result.succeeded()) {
                System.out.println("Server is now listening on port:" + port);
            } else
                System.err.println("Failed to start server:" + result.cause());

        });
    }
}

至此,引入了 RPC 框架的服务提供者模块,已经能够接受请求并完成服务调用了

5.5 代理 (消费者发起调用)

在项目准备阶段,我们已经预留了一段调用服务的代码,只要能够获取到 CatService对象(实现类),就能跑通整个流程。但 CatService的实现类从哪来呢?

总不能把服务提供者的CatServicelmpl复制粘贴到消费者模块吧?要能那样做还需要 RPC 框架干什么?分布式系统中,我们调用其他项目或团队提供的接口时,一般只关注请求参数和响应结果,而不关注具体实现。

在之前的架构中讲过,我们可以通过生成代理对象来简化消费方的调用,代理的实现方式大致分为两类类:静态代理和动态代理,下面依次实现。

5.5.1 静态代理

静态代理是指为每一个特定类型的接口或对象,编写一个代理类。

🌰:在 example-consumer 模块中,创建一个静态代理 CatServiceProxy,实现 CatService接口和 getCatById 方法。

只不过实现 getCatById 方法时,不是复制粘贴服务提供者 CatServicelmpl 中的代码,而是要构造 HTTP 请求去调用服务提供者。

package com.jerry.consumer.proxy;

import cn.hutool.http.HttpRequest;
import cn.hutool.http.HttpResponse;
import com.jerry.aperpc.model.RpcRequest;
import com.jerry.aperpc.model.RpcResponse;
import com.jerry.aperpc.serializer.JDKSerializer;
import com.jerry.aperpc.serializer.KryoSerializer;
import com.jerry.aperpc.serializer.Serializer;
import com.jerry.common.model.Cat;
import com.jerry.common.service.CatService;

import java.io.IOException;

/**
 * @version 1.0
 * @Author jerryLau
 * @Date 2024/4/9 17:14
 * @注释 🐱服务的静态代理
 */

public class CatServiceProxy implements CatService {
    @Override
    public Cat getCat(Cat cat) {
        // 指定序列化器
        final Serializer serializer = new KryoSerializer();

        // 构造请求
        RpcRequest rpcRequest = RpcRequest.builder()
                .serviceName(CatService.class.getName())
                .methodName("getCat")
                .parameterTypes(new Class[]{Cat.class})
                .args(new Object[]{cat})
                .build();
        try {
            // 序列化(Java 对象 => 字节数组)
            byte[] bodyBytes = serializer.serialize(rpcRequest);

            // 发送请求
            try (HttpResponse httpResponse = HttpRequest.post("http://localhost:8080")
                    .body(bodyBytes)
                    .execute()) {
                byte[] result = httpResponse.bodyBytes();
                // 反序列化(字节数组 => Java 对象)
                RpcResponse rpcResponse = serializer.deserialize(result, RpcResponse.class);
                return (Cat) rpcResponse.getData();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }

        return null;
    }

    @Override
    public Cat getCatById(int id) {
        // 指定序列化器
        final Serializer serializer = new KryoSerializer();

        // 构造请求
        RpcRequest rpcRequest = RpcRequest.builder()
                .serviceName(CatService.class.getName())
                .methodName("getCatById")
                .parameterTypes(new Class[]{int.class})
                .args(new Object[]{id})
                .build();
        try {
            // 序列化(Java 对象 => 字节数组)
            byte[] bodyBytes = serializer.serialize(rpcRequest);

            // 发送请求
            try (HttpResponse httpResponse = HttpRequest.post("http://localhost:8080")
                    .body(bodyBytes)
                    .execute()) {
                byte[] result = httpResponse.bodyBytes();
                // 反序列化(字节数组 => Java 对象)
                RpcResponse rpcResponse = serializer.deserialize(result, RpcResponse.class);
                return (Cat) rpcResponse.getData();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }
}

然后修改example-consumer 模块 EasyConsumer,new 一个代理对象并赋值给 CatService,就能完成调用:

package com.jerry.consumer;

import com.jerry.aperpc.proxy.ServiceProxyFactory;
import com.jerry.common.model.Cat;
import com.jerry.common.service.CatService;
import com.jerry.consumer.proxy.CatServiceProxy;

/**
 * @version 1.0
 * @Author jerryLau
 * @Date ${DATE} ${TIME}
 * @注释
 */
public class EasyConsumer {
    public static void main(String[] args) {
        //todo: 需要获取 CatService 的实现类对象
        CatService catService = new CatServiceProxy();

        //调用
        Cat newCat = catService.getCatById(1);
        if (newCat != null) {
            System.out.println("消费者获取到的 cat :" + newCat.toString());
        } else {
            System.out.println("no cat");
        }

    }

}

接下来我们尝试运行一下,优先启动生产者,然后启动消费者,正常的情况下,控制台会出现如下信息

在这里插入图片描述

在这里插入图片描述

静态代理虽然很好理解(就是写个实现类嘛),但缺点也很明显,我们如果要给每个服务接口都写一个实现类,是非常麻烦的,这种代理方式的灵活性也是比较差的。

因此我们尝试使用动态代理

5.5.2 动态代理

动态代理的作用是

根据要生成的对象的类型,自动生成一个代理对象。

常用的动态代理实现方式有JDK动态代理基于字节码生成的动态代理(比如 CGLIB)。前者简单易用、无需引入额外的库,但缺点是只能对接口进行代理;后者更灵活、可以对任何类进行代理,但性能略低于JDK动态代理。

此处我们使用 JDK 动态代理,

  • 在 RPC 模块中编写动态代理类 ServiceProxy,需要实现 InvocationHandler 接口的 invoke 方法

几乎就是将静态代理的方式搬运过来

package com.jerry.aperpc.proxy;

import cn.hutool.http.HttpRequest;
import cn.hutool.http.HttpResponse;
import com.jerry.aperpc.model.RpcRequest;
import com.jerry.aperpc.model.RpcResponse;
import com.jerry.aperpc.serializer.JDKSerializer;
import com.jerry.aperpc.serializer.KryoSerializer;
import com.jerry.aperpc.serializer.Serializer;

import java.io.IOException;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;

/**
 * @version 1.0
 * @Author jerryLau
 * @Date 2024/4/10 8:44
 * @注释 rpc 模块中的动态代理
 */
public class ServiceProxy implements InvocationHandler {

    /***
     * 调用代理
     * @param proxy the proxy instance that the method was invoked on
     *
     * @param method the {@code Method} instance corresponding to
     * the interface method invoked on the proxy instance.  The declaring
     * class of the {@code Method} object will be the interface that
     * the method was declared in, which may be a superinterface of the
     * proxy interface that the proxy class inherits the method through.
     *
     * @param args an array of objects containing the values of the
     * arguments passed in the method invocation on the proxy instance,
     * or {@code null} if interface method takes no arguments.
     * Arguments of primitive types are wrapped in instances of the
     * appropriate primitive wrapper class, such as
     * {@code java.lang.Integer} or {@code java.lang.Boolean}.
     *
     * @return
     * @throws Throwable
     */
    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {

        System.out.println("invoke current  method :" + method + "args: " + args);
        System.out.println("invoke current  method :" + method.getName() + "  args: " + args +"  parameterTypes:"+method.getParameterTypes());
        // 指定序列化器
        final Serializer serializer = new KryoSerializer();

        // 构造请求
        RpcRequest rpcRequest = RpcRequest.builder()
                .serviceName(method.getDeclaringClass().getName())
                .methodName(method.getName())
                .parameterTypes(method.getParameterTypes())
                .args(args)
                .build();

        try {
            // 序列化(Java 对象 => 字节数组)
            byte[] bodyBytes = serializer.serialize(rpcRequest);

            // 发送请求
            //todo:暂时设置成硬编码,后面改成服务发现
            try (HttpResponse httpResponse = HttpRequest.post("http://localhost:8080")
                    .body(bodyBytes)
                    .execute()) {
                byte[] result = httpResponse.bodyBytes();
                // 反序列化(字节数组 => Java 对象)
                RpcResponse rpcResponse = serializer.deserialize(result, RpcResponse.class);
                return rpcResponse.getData();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }

        return null;
    }
}

  • 创建动态代理工厂 ServiceProxyFactory,作用是根据指定类创建动态代理对象

这里是使用了 工厂设计模式,来简化对象的创建过程,代码如下:

package com.jerry.aperpc.proxy;

import java.lang.reflect.Proxy;

/**
 * @version 1.0
 * @Author jerryLau
 * @Date 2024/4/10 8:50
 * @注释 动态代理工厂,通过指定类创建代理对象
 */
public class ServiceProxyFactory {

    public static <T> T getProxy(Class<T> serviceClass) {

        return (T) Proxy.newProxyInstance(
                serviceClass.getClassLoader(),
                new Class[]{serviceClass}
                , new ServiceProxy()
        );
    }

}

上述代码中,主要是通过 Proxy.newProxyInstance 方法为指定类型创建代理对象

  • 最后将example-consumer模块中获取通过静态代理获取CatService,调整成调用动态代理工厂得到动态代理对象
package com.jerry.consumer;

import com.jerry.aperpc.proxy.ServiceProxyFactory;
import com.jerry.common.model.Cat;
import com.jerry.common.service.CatService;
import com.jerry.consumer.proxy.CatServiceProxy;

/**
 * @version 1.0
 * @Author jerryLau
 * @Date ${DATE} ${TIME}
 * @注释
 */
public class EasyConsumer {
    public static void main(String[] args) {
        //todo: 需要获取 CatService 的实现类对象
//        CatService catService = new CatServiceProxy();
        CatService catService = ServiceProxyFactory.getProxy(CatService.class);

        //调用
        Cat newCat = catService.getCatById(1);
        if (newCat != null) {
            System.out.println("消费者获取到的 cat :" + newCat.toString());
        } else {
            System.out.println("no cat");
        }

    }

}

运行结果与静态差不太多

在这里插入图片描述

在这里插入图片描述

简单解释下上述动态代理部分代码执行流程,稍后再测试环节客以debug看到他是如何运行的

当服务消费者中通过调用catService.getCatById时,实际上并没有直接调用 CatService 类的 getCatById 方法,而是调用了 通过动态代理工厂生成的代理对象的 getCatById 方法。这个代理对象内部持有 InvocationHandler 的引用(在这个例子中是 ServiceProxy 的实例),并将方法调用转发给 ServiceProxyinvoke 方法。

invoke 方法是如何拿到对应的方法进行执行的呢?这是通过反射机制实现的。当代理对象的方法被调用时,Proxy 类生成的代理对象会捕获这个调用,并获取被调用的方法(Method 对象)、调用该方法的代理对象本身(proxy 参数)以及传递给该方法的参数(args 参数)。然后,它将这些信息传递给 InvocationHandlerinvoke 方法。

invoke 方法内部,你可以通过 method 参数来获取被调用的方法对象,这个例子中做了系统的打印输出的同时,获取到了要调用的方法信息、传入的参数列表等,这不就是我们服务提供者需要的参数么?用这些参数来构造请求对象就可以完成调用了。

因此,当你通过代理对象调用方法时,实际上执行的是 InvocationHandlerinvoke 方法,而在这个方法中,你可以通过反射机制调用目标对象上的相应方法,并在调用前后添加你想要的额外逻辑。这种方式允许你在不修改目标对象代码的情况下,为目标对象添加一些额外的功能或行为。

需要注意的是:上述代码中,请求的服务提供者地址被硬编码了,后期需要使用注册中心和服务发现机制来解决。

至此,简易版的 RPC 框架已经开发完成,下面我们进行简单的测试。

6、简单测试

  • 以debug模式运行EasyProducer

在这里插入图片描述

  • 以debug模式运行EasyConsumer,在动态代理工厂ServiceProxy设置断点,可以看到调用 catService时,实际是调用了代理对象的 invoke 方法,并且获取到了 serviceName、methodName、参数类型和列表等信息。

在这里插入图片描述

在这里插入图片描述

  • 在服务提供者出请求处理出设置断点,可以看到接受并反序列化后的请求,跟发送时的内容一致

在这里插入图片描述

最后运行输出的结果就在不给大家演示了,参考上面5.5.2 动态代理的运行结果截图


至此,我们实现了简易版的PRC框架

码字不易,希望大家能够一键三连🌝⭐🌟


代码仓库 ape-rpc: 轮子项目,手动实现rpc github🌐 || ape-rpc: 轮子项目,手动实现rpc gitee🌐

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mfbz.cn/a/551713.html

如若内容造成侵权/违法违规/事实不符,请联系我们进行投诉反馈qq邮箱809451989@qq.com,一经查实,立即删除!

相关文章

抖音小店运营计划表年度电商规划管理模板

【干货资料持续更新&#xff0c;以防走丢】 抖音小店运营计划表年度电商规划管理模板 部分资料预览 资料部分是网络整理&#xff0c;仅供学习参考。 抖音店铺运营表格 &#xff08;完整资料包含以下内容&#xff09; 目录 抖音店铺运营计划&#xff1a; 一、店铺定位与目标…

MySql运维篇

目录 一.日志 1.1日志分类 1.2Error Log 1.3BinaryLog 1.4SlowQuery Log 二.备份 2.1备份原因 2.2备份目标 2.3备份技术 2.3.1物理备份 2.3.2逻辑备份 2.4备份方式 2.4.1完全备份 2.4.2增量备份 2.4.3差异备份 2.5备份环境准备 2.6完全备份实验 2.6.1完全备…

书生·浦语大模型全链路开源体系-第4课

书生浦语大模型全链路开源体系-第4课 书生浦语大模型全链路开源体系-第4课相关资源XTuner 微调 LLMXTuner 微调小助手认知环境安装前期准备启动微调模型格式转换模型合并微调结果验证 将认知助手上传至OpenXLab将认知助手应用部署到OpenXLab使用XTuner微调多模态LLM前期准备启动…

连锁服装卖场进销存一般怎么管理

连锁服装卖场的进销存管理是保证业务顺畅运作和最大化利润的关键之一。随着市场竞争的加剧和消费者需求的变化&#xff0c;良好的进销存管理能够帮助企业及时调整库存&#xff0c;减少滞销品&#xff0c;提高资金周转率&#xff0c;从而增强市场竞争力。本文将探讨连锁服装卖场…

单独设置浏览器滚动条上下箭头

解决方法 重点 ::-webkit-scrollbar-button:vertical 给垂直方向的滚动条设置样式 ::-webkit-scrollbar-button:vertical:start 上方向的按钮 ::-webkit-scrollbar-button:vertical:start:decrement 上方向单个按钮 下方向同理 不知道为啥搜索出来的single-button不生效&#…

制造业的数字化转型如何做?

随着科技的迅速发展&#xff0c;数字化转型已经成为制造型企业提高竞争力的关键因素。它可以帮助制造型企业&#xff0c;在产品优化设计、材料采购、生产流程方面实现精细化管理&#xff1b;提升上下游协同生产能力&#xff0c;提高生产效率、降低生产成本、优化产品质量&#…

华为的AI战略地图上,才不是只有大模型

大模型火热了一年&#xff0c;现在还没做AI化改造的企业&#xff0c;就像是工业革命浪潮伊始与火车赛跑的那辆马车。 最早的蒸汽火车缓慢又笨重&#xff0c;甚至铁轨上还预留了马匹行走的空间&#xff0c;以便随时用马拉火车来替代蒸汽火车&#xff0c;一辆华丽的马车试图和火…

浮点数的存储方式、bf16和fp16的区别

目录 1. 小数的二进制转换2. 浮点数的二进制转换3. 浮点数的存储3.1 以fp32为例3.2 规约形式与非规约形式 4. 各种类型的浮点数5. BF16和FP16的区别Ref 1. 小数的二进制转换 十进制小数转换成二进制小数采用「乘2取整&#xff0c;顺序排列」法。具体做法是&#xff1a;用 2 2…

C++语言·类和对象

1. 类的引入 C语言结构体中只能定义变量&#xff0c;但在C中&#xff0c;结构体内不仅可以定义变量&#xff0c;也可以定义函数&#xff0c;同时C中struct的名称就可以代表类型&#xff0c;不用像C那样为了方便还要typedef一下。 在C中我们管定义的结构体类型叫做类(student)&a…

idea 将项目上传到gitee远程仓库具体操作

目录标题 一、新建仓库二、初始化项目三、addcommit四、配置远程仓库五、拉取远程仓库内容六、push代码到仓库 一、新建仓库 新建仓库教程 注意&#xff1a;远程仓库的初始文件不要与本地存在名字一样的文件&#xff0c;不然拉取会因为冲突而失败。可以把远程一样的初始文件删…

汇舟问卷:推荐一个挣外国人钱项目

在海外&#xff0c;问卷调查作为一种普遍的市场研究手段&#xff0c;它们能够为企业下一季度的营销策略调整提供有力的数据支撑。 每份问卷的报酬金额各不相同&#xff0c;最低为1美元&#xff0c;最高可以达到10几美元。大多数问卷的报酬在3到5美元之间。 然而&#xff0c;在…

JS-42-Node.js01-Node.js介绍

一、浏览器大战 众所周知&#xff0c;在Netscape设计出JavaScript后的短短几个月&#xff0c;JavaScript事实上已经是前端开发的唯一标准。 后来&#xff0c;微软通过IE击败了Netscape后一统桌面&#xff0c;结果几年时间&#xff0c;浏览器毫无进步。&#xff08;2001年推出…

HDFS详解(Hadoop)

Hadoop 分布式文件系统&#xff08;Hadoop Distributed File System&#xff0c;HDFS&#xff09;是 Apache Hadoop 生态系统的核心组件之一&#xff0c;它是设计用于存储大规模数据集并运行在廉价硬件上的分布式文件系统。 1. 分布式存储&#xff1a; HDFS 将文件分割成若干块…

【游戏云服务器推荐】幻兽帕鲁 我的世界 雾锁王国 饥荒联机版 英灵神殿通用云服务器 2-64G随意选 附最新价格对比

更新日期&#xff1a;4月17日&#xff08;京东云采购季持续进行&#xff09; 本文纯原创&#xff0c;侵权必究 《最新对比表》已更新在文章头部—腾讯云文档&#xff0c;文章具有时效性&#xff0c;请以腾讯文档为准&#xff01; 【腾讯文档实时更新】2024年-幻兽帕鲁服务器专…

李飞飞团队发布《2024年人工智能指数报告》,预测人工智能未来发展趋势

昨天&#xff0c;斯坦福大学 Human-Center Artificial Intelligence (HAI)研究中心发布了《2024年人工智能指数报告》。 由斯坦福大学发起的人工智能指数&#xff08;AI Index&#xff09;是一个追踪 AI 动态和进展的非营利性项目&#xff0c;旨在全面研究 AI 行业状况&#xf…

不同质量图在卡尔曼滤波相位解缠中应用探讨

文献来源&#xff1a;不同质量图在卡尔曼滤波相位解缠中应用探讨 闫 满&#xff0c;郭春华 测绘科学技术, 2019, 7(2), 65-73 卡尔曼滤波将相位解缠转化为状态估计问题&#xff0c;实现相位解缠与噪声消除的一并处理。通过建立相位的动 态方程和观测方程来求解真实相位&#x…

The O-one:开源语言模型计算机的革命

在人工智能的浪潮中&#xff0c;The O-one作为一个创新的开源项目&#xff0c;正以其独特的功能和开放性吸引着全球开发者和科技爱好者的目光。这个项目不仅仅是一个简单的语言模型&#xff0c;它是一个能够通过语音交互与计算机进行对话的智能系统&#xff0c;极大地提升了人机…

Java面试题笔记(持续更新)

Java基础 java中的Math.round(-1.5)等于多少&#xff1f; Math的round方法是四舍五入,如果参数是负数,则往大的数如,Math.round(-1.5)-1&#xff0c;如果是Math.round(1.5)则结果为2 JDK和JRE的区别&#xff1f; JDK 是 Java Development ToolKit 的简称&#xff0c;也就是…

Java——代码块

目录 一.代码块概念以及分类 二.普通代码块 三.构造代码块 四.静态代码块 一.代码块概念以及分类 使用 {} 定义的一段代码称为代码块。根据代码块定义的位置以及关键字&#xff0c;又可分为以下四种&#xff1a; 普通代码块构造块静态块同步代码块&#xff08;后续讲解多…

MySQL数据库第二天

如何授权和撤销 已经给客户授权&#xff1a; GRANT all on *.* to "用户名""获取IP地址" identified by "密码" 如果想撤销可以使用&#xff1a; revoke all on 数据库.表 form "用户名""获取的IP地址" 补充&#xff1…
最新文章