徒手擼一個簡單的RPC框架

徒手擼一個簡單的RPC框架

以前在牛逼哄哄的 RPC 框架，底層到底什麼原理得知了RPC（遠程過程調用）簡單來講就是調用遠程的服務就像調用本地方法同樣，其中用到的知識有序列化和反序列化、動態代理、網絡傳輸、動態加載、反射這些知識點。發現這些知識都瞭解一些。因此就想着試試本身實現一個簡單的RPC框架，即鞏固了基礎的知識，也能更加深刻的瞭解RPC原理。固然一個完整的RPC框架包含了許多的功能，例如服務的發現與治理，網關等等。本篇只是簡單的實現了一個調用的過程。

傳參出參分析

一個簡單請求能夠抽象爲兩步

那麼就根據這兩步進行分析，在請求以前咱們應該發送給服務端什麼信息？而服務端處理完之後應該返回客戶端什麼信息？

在請求以前咱們應該發送給服務端什麼信息？

因爲咱們在客戶端調用的是服務端提供的接口，因此咱們須要將客戶端調用的信息傳輸過去，那麼咱們能夠將要傳輸的信息分爲兩類

• 第一類是服務端能夠根據這個信息找到相應的接口實現類和方法
• 第二類是調用此方法傳輸的參數信息

那麼咱們就根據要傳輸的兩類信息進行分析，什麼信息可以找到相應的實現類的相應的方法？要找到方法必需要先找到類，這裏咱們能夠簡單的用Spring提供的Bean實例管理ApplicationContext進行類的尋找。因此要找到類的實例只須要知道此類的名字就行，找到了類的實例，那麼如何找到方法呢？在反射中經過反射可以根據方法名和參數類型從而找到這個方法。那麼此時第一類的信息咱們就明瞭了，那麼就創建相應的是實體類存儲這些信息。

@Data
public class Request implements Serializable {
    private static final long serialVersionUID = 3933918042687238629L;
    private String className;
    private String methodName;
    private Class<?> [] parameTypes;
    private Object [] parameters;
}

服務端處理完之後應該返回客戶端什麼信息？

上面咱們分析了客戶端應該傳輸什麼信息給服務端，那麼服務端處理完之後應該傳什麼樣的返回值呢？這裏咱們只考慮最簡單的狀況，客戶端請求的線程也會一直在等着，不會有異步處理這一說，因此這麼分析的話就簡單了，直接將獲得的處理結果返回就好了。

@Data
public class Response implements Serializable {
    private static final long serialVersionUID = -2393333111247658778L;
    private Object result;
}

因爲都涉及到了網絡傳輸，因此都要實現序列化的接口

如何得到傳參信息並執行？-客戶端

上面咱們分析了客戶端向服務端發送的信息都有哪些？那麼咱們如何得到這些信息呢？首先咱們調用的是接口，因此咱們須要寫自定義註解而後在程序啓動的時候將這些信息加載在Spring容器中。有了這些信息那麼咱們就須要傳輸了，調用接口可是實際上執行的確實網絡傳輸的過程，因此咱們須要動態代理。那麼就能夠分爲如下兩步

• 初始化信息階段：將key爲接口名，value爲動態接口類註冊進Spring容器中
• 執行階段：經過動態代理，實際執行網絡傳輸

初始化信息階段

因爲咱們使用Spring做爲Bean的管理，因此要將接口和對應的代理類註冊進Spring容器中。而咱們如何找到咱們想要調用的接口類呢？咱們能夠自定義註解進行掃描。將想要調用的接口所有註冊進容器中。

建立一個註解類，用於標註哪些接口是能夠進行Rpc的

@Target({ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
public @interface RpcClient {
}

而後建立對於@RpcClient註解的掃描類RpcInitConfig，將其註冊進Spring容器中

public class RpcInitConfig implements ImportBeanDefinitionRegistrar{
    
    @Override
    public void registerBeanDefinitions(AnnotationMetadata importingClassMetadata, BeanDefinitionRegistry registry) {
        ClassPathScanningCandidateComponentProvider provider = getScanner();
        //設置掃描器
        provider.addIncludeFilter(new AnnotationTypeFilter(RpcClient.class));
        //掃描此包下的全部帶有@RpcClient的註解的類
        Set<BeanDefinition> beanDefinitionSet = provider.findCandidateComponents("com.example.rpcclient.client");
        for (BeanDefinition beanDefinition : beanDefinitionSet){
            if (beanDefinition instanceof AnnotatedBeanDefinition){
                //得到註解上的參數信息
                AnnotatedBeanDefinition annotatedBeanDefinition = (AnnotatedBeanDefinition) beanDefinition;
                String beanClassAllName = beanDefinition.getBeanClassName();
                Map<String, Object> paraMap = annotatedBeanDefinition.getMetadata()
                        .getAnnotationAttributes(RpcClient.class.getCanonicalName());
                //將RpcClient的工廠類註冊進去
                BeanDefinitionBuilder builder = BeanDefinitionBuilder
                        .genericBeanDefinition(RpcClinetFactoryBean.class);
                //設置RpcClinetFactoryBean工廠類中的構造函數的值
                builder.addConstructorArgValue(beanClassAllName);
                builder.getBeanDefinition().setAutowireMode(AbstractBeanDefinition.AUTOWIRE_BY_TYPE);
                //將其註冊進容器中
                registry.registerBeanDefinition(
                        beanClassAllName ,
                        builder.getBeanDefinition());
            }
        }
    }
    //容許Spring掃描接口上的註解
    protected ClassPathScanningCandidateComponentProvider getScanner() {
        return new ClassPathScanningCandidateComponentProvider(false) {
            @Override
            protected boolean isCandidateComponent(AnnotatedBeanDefinition beanDefinition) {
                return beanDefinition.getMetadata().isInterface() && beanDefinition.getMetadata().isIndependent();
            }
        };
    }
}

因爲上面註冊的是工廠類，因此咱們創建一個工廠類RpcClinetFactoryBean繼承Spring中的FactoryBean類，由其統一建立@RpcClient註解的代理類

@Data
public class RpcClinetFactoryBean implements FactoryBean {

    @Autowired
    private RpcDynamicPro rpcDynamicPro;

    private Class<?> classType;


    public RpcClinetFactoryBean(Class<?> classType) {
        this.classType = classType;
    }

    @Override
    public Object getObject(){
        ClassLoader classLoader = classType.getClassLoader();
        Object object = Proxy.newProxyInstance(classLoader,new Class<?>[]{classType},rpcDynamicPro);
        return object;
    }

    @Override
    public Class<?> getObjectType() {
        return this.classType;
    }

    @Override
    public boolean isSingleton() {
        return false;
    }
}

注意此處的 getObjectType 方法，在將工廠類注入到容器中的時候，這個方法返回的是什麼Class類型那麼註冊進容器中就是什麼Class類型。

而後看一下咱們建立的代理類rpcDynamicPro

@Component
@Slf4j
public class RpcDynamicPro implements InvocationHandler {

    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
       String requestJson = objectToJson(method,args);
        Socket client = new Socket("127.0.0.1", 20006);
        client.setSoTimeout(10000);
        //獲取Socket的輸出流，用來發送數據到服務端
        PrintStream out = new PrintStream(client.getOutputStream());
        //獲取Socket的輸入流，用來接收從服務端發送過來的數據
        BufferedReader buf =  new BufferedReader(new InputStreamReader(client.getInputStream()));
        //發送數據到服務端
        out.println(requestJson);
        Response response = new Response();
        Gson gson =new Gson();
        try{
            //從服務器端接收數據有個時間限制（系統自設，也能夠本身設置），超過了這個時間，便會拋出該異常
            String responsJson = buf.readLine();
            response = gson.fromJson(responsJson, Response.class);
        }catch(SocketTimeoutException e){
            log.info("Time out, No response");
        }
        if(client != null){
            //若是構造函數創建起了鏈接，則關閉套接字，若是沒有創建起鏈接，天然不用關閉
            client.close(); //只關閉socket，其關聯的輸入輸出流也會被關閉
        }
        return response.getResult();
    }

    public String objectToJson(Method method,Object [] args){
        Request request = new Request();
        String methodName = method.getName();
        Class<?>[] parameterTypes = method.getParameterTypes();
        String className = method.getDeclaringClass().getName();
        request.setMethodName(methodName);
        request.setParameTypes(parameterTypes);
        request.setParameters(args);
        request.setClassName(getClassName(className));
        GsonBuilder gsonBuilder = new GsonBuilder();
        gsonBuilder.registerTypeAdapterFactory(new ClassTypeAdapterFactory());
        Gson gson = gsonBuilder.create();
        return gson.toJson(request);
    }

    private String getClassName(String beanClassName){
        String className = beanClassName.substring(beanClassName.lastIndexOf(".")+1);
        className = className.substring(0,1).toLowerCase() + className.substring(1);
        return className;
    }
}

咱們的客戶端已經寫完了，傳給服務端的信息咱們也已經拼裝完畢了。剩下的工做就簡單了，開始編寫服務端的代碼。

服務端處理完之後應該返回客戶端什麼信息？-服務端

服務端的代碼相比較客戶端來講要簡單一些。能夠簡單分爲下面三步

• 拿到接口名之後，經過接口名找到實現類
• 經過反射進行對應方法的執行
• 返回執行完的信息

那麼咱們就根據這三步進行編寫代碼

拿到接口名之後，經過接口名找到實現類

如何經過接口名拿到對應接口的實現類呢？這就須要咱們在服務端啓動的時候將其對應信息加載進去

@Component
@Log4j
public class InitRpcConfig implements CommandLineRunner {
    @Autowired
    private ApplicationContext applicationContext;

    public static Map<String,Object> rpcServiceMap = new HashMap<>();

    @Override
    public void run(String... args) throws Exception {
        Map<String, Object> beansWithAnnotation = applicationContext.getBeansWithAnnotation(Service.class);
        for (Object bean: beansWithAnnotation.values()){
            Class<?> clazz = bean.getClass();
            Class<?>[] interfaces = clazz.getInterfaces();
            for (Class<?> inter : interfaces){
                rpcServiceMap.put(getClassName(inter.getName()),bean);
                log.info("已經加載的服務:"+inter.getName());
            }
        }
    }

    private String getClassName(String beanClassName){
        String className = beanClassName.substring(beanClassName.lastIndexOf(".")+1);
        className = className.substring(0,1).toLowerCase() + className.substring(1);
        return className;
    }
}

此時rpcServiceMap 存儲的就是接口名和其對應的實現類的對應關係。

經過反射進行對應方法的執行

此時拿到了對應關係之後就能根據客戶端傳過來的信息找到相應的實現類中的方法。而後進行執行並返回信息就行

public Response invokeMethod(Request request){
        String className = request.getClassName();
        String methodName = request.getMethodName();
        Object[] parameters = request.getParameters();
        Class<?>[] parameTypes = request.getParameTypes();
        Object o = InitRpcConfig.rpcServiceMap.get(className);
        Response response = new Response();
        try {
            Method method = o.getClass().getDeclaredMethod(methodName, parameTypes);
            Object invokeMethod = method.invoke(o, parameters);
            response.setResult(invokeMethod);
        } catch (NoSuchMethodException e) {
            log.info("沒有找到"+methodName);
        } catch (IllegalAccessException e) {
            log.info("執行錯誤"+parameters);
        } catch (InvocationTargetException e) {
            log.info("執行錯誤"+parameters);
        }
        return response;
    }

如今咱們兩個服務都啓動起來而且在客戶端進行調用就發現只是調用接口就能調用過來了。

總結

到如今一個簡單的RPC就完成了，可是其中還有不少的功能須要完善，例如一個完整RPC框架確定還須要服務註冊與發現，並且雙方通訊確定也不能是直接開啓一個線程一直在等着，確定須要是異步的等等的各類功能。後面隨着學習的深刻，這個框架也會慢慢增長一些東西。不只是對所學知識的一個應用，更是一個總結。有時候學一個東西學起來以爲很簡單，可是真正應用的時候就會發現各類各樣的小問題。好比在寫這個例子的時候碰到一個問題就是@Autowired的時候一直找不到SendMessage的類型，最後才發現是工廠類RpcClinetFactoryBean 中的getObjectType 中的返回類型寫錯了，我以前寫的是

public Class<?> getObjectType() {
        return this.getClass();;
    }

這樣的話註冊進容器的就是RpcClinetFactoryBean 類型的而不是SendMessage 的類型。

完整項目地址