10 个类手写实现 RPC 通信框架

青旅半醒 2021-09-01 14:32 378阅读 0赞

点击上方 [Java后端][Java]，选择 设为星标

优质文章，及时送达

--------------------

作者：autu

链接：autumn200.com

**什么是rpc**

RPC：remote procedure call Protocol 远程过程调用 调用远程服务，就像调用本地的服务一样，不用关心调用细节，就像调用本机的服务一样的

### **RPC原理** ###

实现RPC通信的程序包括5个部分：rpc-client、客户端proxy、socket、服务端proxy、rpc-server

![format_png][]

#### **request** ####

*  客户端：当rpc-client发起远程调用时，实际上是通过客户端代理 将要调用的接口、方法、参数、参数类型进行**序列化**，然后通过socket实时将封装调用参数的实例发送到服务端。
 *  服务端：socket接收到客户端传来的信息进行反序列化，然后通过这些信息委派到具体的实现对象

#### **response** ####

*  服务端：目标方法执行完成后，将执行结果返回给socket
 *  客户端：socket接收到结果后，返回给rpc-client，调用结束

### **应用到的技术** ###

*  java
 *  spring
 *  序列化
 *  socket
 *  反射
 *  动态代理

### **项目GitHub地址** ###

https://github.com/autumnqfeng/write\_rpc

### **掘金地址** ###

https://juejin.im/post/5eef0fe551882565a459e8b6

### **服务端项目** ###

#### 项目结构 ####

rpc-server项目包含2个子项目：order-api、order-provider

order-api中存放请求接口与`RpcRequest`（类名、方法名、参数的实体类）

order-provider为请求接口实现、socket、proxy相关类

![format_png 1][]

##### order-api #####

![format_png 2][]

##### order-provider #####

![format_png 3][]

#### **服务注册** ####

要想实现动态调用`ServiceImpl`，关键就需要将service类管理起来，那问题来了，我们如何管理这些服务类呢？

我们可以参照spring中的`@Service`注解，通过自定义注解的方式来做到服务注册，我们定义一个注解`@RpcRemoteService`作用在`ServiceImpl`类上，将标记此注解的类名、方法名保存到Map中，以此来定位到具体实现类。

##### `@RpcRemoteService`注解 #####

/**
     * 服务端服务发现注解
     *
     * @author: ***
     * @date: 2020/6/21 16:21
     */
    @Target(ElementType.TYPE)
    @Retention(RetentionPolicy.RUNTIME)
    @Component
    public @interface RpcRemoteService {
    }

##### 服务注册类`InitialMerdiator` #####

在spring容器初始化完成之后，扫描到`@RpcRemoteService`标记的类，并保存到`Mediator.ROUTING`中。

/**
     * 初始化中间代理层对象
     *
     * @author: ***
     * @date: 2020/6/21 16:33
     */
    @Component
    public class InitialMerdiator implements BeanPostProcessor {
    
        @Override
        public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
            //加了服务发布标记的bean进行远程发布
            if (bean.getClass().isAnnotationPresent(RpcRemoteService.class)) {
                Method[] methods = bean.getClass().getDeclaredMethods();
                for (Method method : methods) {
                    String routingKey = bean.getClass().getInterfaces()[0].getName() + "." + method.getName();
                    BeanMethod beanMethod = new BeanMethod();
                    beanMethod.setBean(bean);
                    beanMethod.setMethod(method);
                    Mediator.ROUTING.put(routingKey, beanMethod);
                }
            }
            return bean;
        }
    }

**socket监听**

socket监听客户端请求

##### socket启动类`SocketServer` #####

spring容器加载完成之后，启动socket

/**
     * spring 容器启动完成之后，会发布一个ContextRefreshedEven
     * 容器启动后启动socket监听
     *
     * @author: ***
     * @date: 2020/6/21 16:51
     */
    @Component
    public class SocketServer implements ApplicationListener<ContextRefreshedEvent> {
        private final ExecutorService executorService= Executors.newCachedThreadPool();
    
        @Override
        public void onApplicationEvent(ContextRefreshedEvent contextRefreshedEvent) {
            ServerSocket serverSocket=null;
            try {
                serverSocket = new ServerSocket(8888);
                while (true) {
                    Socket accept = serverSocket.accept();
                    // 线程池处理socket
            executorService.execute(new ProcessorHandler(accept));
                }
            } catch (IOException e) {
                e.printStackTrace();
            } finally {
                if (serverSocket != null) {
                    try {
                        serverSocket.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }
        }

socket处理类`ProcessorHandler`

处理监听到的每个socket

public class ProcessorHandler implements Runnable {
    
        private Socket socket;
    
        public ProcessorHandler(Socket socket) {
            this.socket = socket;
        }
    
        @Override
        public void run() {
            ObjectInputStream inputStream = null;
            ObjectOutputStream outputStream = null;
            try {
                inputStream = new ObjectInputStream(socket.getInputStream());
                // 反序列化
                RpcRequest rpcRequest = (RpcRequest) inputStream.readObject();
    
                // 中间代理执行目标方法
                Mediator mediator = Mediator.getInstance();
                Object response = mediator.processor(rpcRequest);
                System.out.println("服务端的执行结果："+response);
    
                outputStream = new ObjectOutputStream(socket.getOutputStream());
                outputStream.writeObject(response);
                outputStream.flush();
    
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                closeStream(inputStream, outputStream);
            }
        }
    
        private void closeStream(ObjectInputStream inputStream, ObjectOutputStream outputStream) {
            // 关闭流
            if(inputStream!=null){
                try {
                    inputStream.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            if (outputStream!=null){
                try {
                    outputStream.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }

**服务端代理**

##### `Mediator` #####

/**
     * 服务端socket与目标方法的中间代理层
     *
     * @author: ***
     * @date: 2020/6/21 16:25
     */
    public class Mediator {
    
        /** 用来存储发布的服务的实例(服务调用的路由) */
        public static Map<String, BeanMethod> ROUTING = new ConcurrentHashMap<>();
    
        /** 单例模式创建该代理层实例 */
        private volatile static Mediator instance;
    
        private Mediator() {
        }
    
        public static Mediator getInstance() {
            if (instance == null) {
                synchronized (Mediator.class) {
                    if (instance == null) {
                        instance = new Mediator();
                    }
                }
            }
            return instance;
        }
    
        public Object processor(RpcRequest rpcRequest) {
            // 路由key
            String routingKey = rpcRequest.getClassName() + "." + rpcRequest.getMethodName();
            BeanMethod beanMethod = ROUTING.get(routingKey);
            if (beanMethod == null) {
                return null;
            }
            // 执行目标方法
            Object bean = beanMethod.getBean();
            Method method = beanMethod.getMethod();
            try {
                return method.invoke(bean, rpcRequest.getArgs());
            } catch (Exception e) {
                e.printStackTrace();
            }
            return null;
        }
    }

##### BeanMethod #####

/**
     * 中间层反射调用时，存储目标方法、目标类的实体
     *
     * @author: ***
     * @date: 2020/6/21 16:43
     */
    public class BeanMethod {
    
        private Object bean;
    
        private Method method;
    
        // setter、getter略
    }

### **客户端项目**    ###

#### 项目结构 ####

![format_png 4][]

#### **服务发现** ####

服务发现我们同样使用注解来做，这就需要参照Spring中`@Autowired`的原理实现，我们自定义`@RpcReference`注解，定义在字段上，将接口实现的**代理类**注入到该字段上。

##### `@RpcReference`注解 #####

/**
     * 服务注入注解
     *
     * @author: ***
     * @date: 2020/6/20 22:41
     */
    @Target(ElementType.FIELD)
    @Retention(RetentionPolicy.RUNTIME)
    @Component
    public @interface RpcReference {

##### 服务发现类`ReferenceInvokeProxy` #####

在spring容器初始化之前，扫描bean中所有`@RpcReference`注解标记的字段。

/**
     * 远程动态调用service代理
     *
     * @author: ***
     * @date: 2020/6/20 22:44
     */
    @Component
    public class ReferenceInvokeProxy implements BeanPostProcessor {
    
        @Autowired
        private RemoteInvocationHandler invocationHandler;
    
        @Override
        public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
            Field[] fields = bean.getClass().getDeclaredFields();
            for (Field field : fields) {
                if (field.isAnnotationPresent(RpcReference.class)) {
                    field.setAccessible(true);
                    // 针对这个加了RpcReference注解的字段，设置为一个代理的值
                    Object proxy = Proxy.newProxyInstance(field.getType().getClassLoader(), new Class<?>[]{field.getType()}, invocationHandler);
                    try {
                        // 相当于针对加了RpcReference的注解，设置了一个代理，这个代理的实现是invocationHandler
                        field.set(bean, proxy);
                    } catch (IllegalAccessException e) {
                        e.printStackTrace();
                    }
                }
            }
            return bean;
        }
    }

**客户端代理**

##### 客户端动态代理`InvocationHandler`实现类`RemoteInvocationHandler` #####

将目标方法名、目标类名、参数信息封装到`RpcRequest`，然后交给socket发送到服务端。

/**
     * @author: ***
     * @date: 2020/6/20 22:51
     */
    @Component
    public class RemoteInvocationHandler implements InvocationHandler {
    
        @Autowired
        private RpcNetTransport rpcNetTransport;
    
        @Override
        public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
            RpcRequest rpcRequest = new RpcRequest();
            rpcRequest.setClassName(method.getDeclaringClass().getName());
            rpcRequest.setMethodName(method.getName());
            rpcRequest.setTypes(method.getParameterTypes());
            rpcRequest.setArgs(args);
            return rpcNetTransport.send(rpcRequest);
        }
    }

**客户端socket**

##### 网络传输`RpcNetTransport` #####

/**
     * rpc socket 网络传输
     *
     * @author: ***
     * @date: 2020/6/20 22:59
     */
    @Component
    public class RpcNetTransport {
        @Value("${rpc.host}")
        private String host;
        @Value("${rpc.port}")
        private int port;
    
    
        public Object send(RpcRequest rpcRequest) {
            ObjectOutputStream outputStream = null;
            ObjectInputStream inputStream = null;
            try {
                Socket socket = new Socket(host, port);
                // 发送目标方法信息
                outputStream = new ObjectOutputStream(socket.getOutputStream());
                outputStream.writeObject(rpcRequest);
                outputStream.flush();
          // 接收返回值
                inputStream = new ObjectInputStream(socket.getInputStream());
                return inputStream.readObject();
            } catch (IOException | ClassNotFoundException e) {
                e.printStackTrace();
            } finally {
                closeStream(inputStream, outputStream);
            }
            return null;
        }
    
        private void closeStream(ObjectInputStream inputStream, ObjectOutputStream outputStream) {
            // 关闭流
            if(inputStream!=null){
                try {
                    inputStream.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            if (outputStream!=null){
                try {
                    outputStream.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }

--------------------

\-END-

如果看到这里，说明你喜欢这篇文章，请 转发、点赞。同时 标星（置顶）本公众号可以第一时间接受到博文推送。

**推****荐****阅****读**

***1. ***[不会 Java 注解 ? 看这一篇文章！][Java _ _]

***2. ***[线程异常如何处理？试试这样][Link 1]

***3.*** [采用 Vue 编写的功能强大的 Swagger-ui 页面][Vue _ Swagger-ui]

***4. ***[Spring Boot+Redis实现接口自动幂等][Spring Boot_Redis]

最近整理一份面试资料**《Java技术栈学习手册》**，覆盖了Java技术、面试题精选、Spring全家桶、Nginx、SSM、微服务、数据库、数据结构、架构等等。

获取方式：点“ **在看**，关注公众号 **Java后端** 并回复 **777** 领取，更多内容陆续奉上。

喜欢文章，点个在看 ![format_png 5][]

[Java]: http://mp.weixin.qq.com/s?__biz=Mzg2MjEwMjI1Mg%3D%3D&chksm=ce0da17bf97a286df119047a25b7de51621fe542816673bb4aa3ebd8ab97b09a86ed7db2e2ee&idx=3&mid=2247489272&scene=21&sn=ce74938072fc73a808d83ccb474db46d#wechat_redirect
[format_png]: /images/20210730/f32b782218ea4bb098259cf2547f52a5.png
[format_png 1]: /images/20210730/08f3508daf96477b9061c01d2f5ac809.png
[format_png 2]: /images/20210730/1b92c234716044c397e4998c51e8417c.png
[format_png 3]: /images/20210730/a9736257d7384e79a959e097545b2669.png
[format_png 4]: /images/20210730/dc6e0bc1de0d4ea3a4751db387217c85.png
[Java _ _]: http://mp.weixin.qq.com/s?__biz=Mzg2MjEwMjI1Mg%3D%3D&chksm=ce0e4e23f979c735e24958ee242d3a958a87838f98d8c9b030de6770b9192fd693c0d3d48f3e&idx=2&mid=2247498144&scene=21&sn=78c018b992c7d8ef1ea6910e6c01dc31#wechat_redirect
[Link 1]: http://mp.weixin.qq.com/s?__biz=Mzg2MjEwMjI1Mg%3D%3D&chksm=ce0e4e23f979c7354e6c7f0a4c49e3352a9a8cdd77fcfe93c7ce3d1932fcb6f95bcce49251fa&idx=3&mid=2247498144&scene=21&sn=6c98c5f8f957feff17fc5e6cf1a5adaa#wechat_redirect
[Vue _ Swagger-ui]: http://mp.weixin.qq.com/s?__biz=Mzg2MjEwMjI1Mg%3D%3D&chksm=ce0e4e0af979c71c7de64873ef6dc753853761b6d3152f2d4ec4ca0cf66a52a44f17e0c38727&idx=1&mid=2247498121&scene=21&sn=1e74ad95f4f83ed36ba83a5dd933269f#wechat_redirect
[Spring Boot_Redis]: http://mp.weixin.qq.com/s?__biz=Mzg2MjEwMjI1Mg%3D%3D&chksm=ce0e517ef979d868e4f33b739a9f02483f91af2c6337b1e8d899c97f6df2556592a85a9d43ad&idx=1&mid=2247493373&scene=21&sn=6953f10125945ad6105d603cf0750d6b#wechat_redirect
[format_png 5]: /images/20210730/ae25e0bb649f48a5b8c1afbee4ed0d47.png