学会RPC原理这一篇就够了

前言

学习RPC需要会的前置知识：基于TCP的Socket、Java反射的基本使用、序列化、代理模式之动态代理（JDK就行）、多线程，都是一些Java基础知识，不会的自行补课。

什么是 RPC 框架

RPC 框架—– 远程过程调用协议RPC（Remote Procedure Call Protocol)—–允许像调用本地服务一样调用远程服务。

言：RPC知识一种框架思想，A端等接收到B端的序列化数据，在处理好后，再能序列化给B端，我们就可以称它是PRC，以下知识只我个人对PRC的某一种实现方式理解

简单的来说就是我这台服务器A（注册中心）要调服务器B的业务，服务器B有具体的执行逻辑，而服务器A只需要知道接口方法就行。

为什么服务器A只需要知道接口方法，中间在哪用到了动态代理、反射、Socket技术等技术

服务器A执行的方法是实际是动态代理对象在执行，这时我们就可以用代理对象InvocationHandler中的invoke（）方法，这个方法可以知道当前对象正执行的方法名和方法参数，
在通过这个方法以及得到的信息去使用ServerSocket+序列化发送给服务器B，让B接收一些class名、方法名、方法参数等
服务器B在拿到这些参数就可以利用反射技术执行具体逻辑了，然后再将执行后结果继续使用socket的方式返回给A。（为了避免socket接收后阻塞，可以用多线程接收）

RPC 调用的基本流程

1，远程服务之间建立通讯协议

2，寻址：服务器（如主机或IP地址）以及特定的端口，方法的名称名称是什么

3，通过序列化和反序列化进行数据传递

4，将传递过来的数据通过java反射原理定位接口方法和参数

5，暴露服务：用map将寻址的信息暴露给远方服务（提供一个endpoint URI或者一个前端展示页面）

6，多线程并发请求业务

代码实现

服务器A (注册中心），为什么叫注册中心，类似SpringClond服务发现、服务注册功能，把具体接口的实现类用一个map 装起来，供服务B发送接口方式时，服务A好利用反射实现。
主要任务是
1）注册供远程调用的实现类
2）等待服务器B连接
3）反射出具体对象，并执行方法
4）发送处理结果

// 规范接口
public interface Server {

    public void start() throws IOException;

    public void  stop();

    public void register(Class service, Class serviceImpl);
}

/**====================== **/

import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @version 1.0
 * @date 2022/5/5 10:01
 * @desc 注册中心
 */
public class ServerCenter implements Server{

    private static int port;
    private static Boolean isRunning = false;
    // key : 接口名，value ：接口实现类
    private static Map<String,Class> serviceRegister = new HashMap<>();
    // 连接池
    private static ExecutorService  executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

    public ServerCenter(int port){
        this.port = port;
    }

    @Override
    public void start() throws IOException {
        ServerSocket serverSocket = new ServerSocket();
        serverSocket.bind(new InetSocketAddress(port));
        isRunning = true;
        System.out.println("服务器启动完毕");
        while (isRunning){
            Socket socket = serverSocket.accept();// 等待客户端连接 
            // 接收到一个连接，就交给一个线程去处理具体业务
            executor.execute(new Task(socket));
        }
    }

    @Override
    public void stop() {
        isRunning = false;
        executor.shutdown();
        System.out.println("服务器关闭");
    }

    /**
     * @param service 接口的类
     * @param serviceImpl 接口的实现类
     */
    @Override
    public void register(Class service, Class serviceImpl) {
        serviceRegister.put(service.getName(),serviceImpl);
    }
    
	
    class Task implements Runnable{
        Socket socket;
        public Task(Socket socket) {
            this.socket = socket;
        }

        @Override
        public void run() {
            ObjectOutputStream output = null;
            ObjectInputStream input = null;
            try {

                // 接收用户请求
                input = new ObjectInputStream( socket.getInputStream());
                String serviceName = input.readUTF();  // 接口名
                String methodName = input.readUTF();   // 方法名
                Class[] parameterTypes = (Class[])input.readObject(); // 参数类型
                Object[] arguments = (Object[])input.readObject(); // 参数

                // 去注册中心serviceRegister拿具体接口,反射实现类
                Class serviceClass = serviceRegister.get(serviceName);
                Method method = serviceClass.getMethod(methodName, parameterTypes);
                Object result = method.invoke(serviceClass.newInstance(), arguments);

                // 回馈用户
                output = new ObjectOutputStream(socket.getOutputStream());
                output.writeObject(result);
            } catch (Exception exception) {
                exception.printStackTrace();
            }  finally {
                try {
                    output.close();
                    input.close();
                } catch (IOException exception) {
                    exception.printStackTrace();
                }
            }
        }
    }
}

服务B客户端，主要任务是
1）与服务器A建立连接
2）将接口方法和参数等信息发送给服务器A
3）等待接收处理结果


import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.net.InetSocketAddress;
import java.net.Socket;

/**
 * @version 1.0
 * @date 2022/5/5 10:12
 * @desc 客户端， 动态代理
 */
public class Client {

    //  获取代表服务端接口的动态代理对象

    /**
     *
     * @param serviceInterface 请求接口名
     * @param address ip和端口
     * @param <T>
     */
    public static <T> T getRemoteProxyObj(Class serviceInterface , InetSocketAddress address){
        /**
         * Proxy.newProxyInstance(a,b,c)
         * a ： 类加载器，需要代理哪个类
         * b ： 需要代理的对象，具备哪些方法
         * c ： 动态代理的对象
         */
        return (T) Proxy.newProxyInstance(serviceInterface.getClassLoader(), new Class<?>[]{serviceInterface}, new InvocationHandler() {
            // proxy ：newProxyInstance返回代理的对象  mothod：代理对象的哪个方法： args：参数
            @Override
            public Object invoke(Object proxy, Method method, Object[] args) {
                Socket socket =new Socket();
                ObjectOutputStream outputStream = null;
                ObjectInputStream input = null;
                try {
                    socket.connect(address);

                    outputStream = new ObjectOutputStream(socket.getOutputStream());

                    // 顺序自身定义 - 告诉注册中心你要执行哪些方法
                    // 接口名、方法名
                    outputStream.writeUTF(serviceInterface.getName());
                    outputStream.writeUTF(method.getName());
                    // 方法参数类型 方法参数
                    outputStream.writeObject(method.getParameterTypes());
                    outputStream.writeObject(args);

                    // 等待服务端
                    // 接收服务的处理后的值
                    input = new ObjectInputStream(socket.getInputStream());
                    return input.readObject();
                } catch (Exception exception) {
                    exception.printStackTrace();
                    return null;
                } finally {
                    try {
                        outputStream.close();
                        input.close();
                    } catch (IOException exception) {
                        exception.printStackTrace();
                    }
                }
            }
        });
    }
}

Service

public interface RpcService {
    String getName(String name);
}
/**===========**/
public class RpcServiceImpl implements RpcService{
    @Override
    public String getName(String name) {
        return name + " 你好呀！";
    }
}

模拟服务端和客户端

public class RPCServerTest {
    public static void main(String[] args) throws IOException, InterruptedException {
        Server server = new ServerCenter(9999);
        server.register(RpcService.class, RpcServiceImpl.class); //服务注册

        server.start();
    }
}

public class RPCClientTest {
    public static void main(String[] args) {
        RpcService remoteProxyObj = new Client().getRemoteProxyObj(RpcService.class, new InetSocketAddress("127.0.0.1", 9999));

        System.out.println(remoteProxyObj.getName("zs"));
    }

}

文章由极客之音整理，本文链接：https://www.bmabk.com/index.php/post/4804.html

学会RPC原理这一篇就够了

前言

什么是 RPC 框架

RPC 调用的基本流程

代码实现

相关推荐