从零开始，徒手撸一个简单的 RPC 框架，轻松搞定！

2021-05-06 535

得知了RPC（远程过程调用）简单来说就是调用远程的服务就像调用本地方法一样，其中用到的知识有序列化和反序列化、动态代理、网络传输、动态加载、反射这些知识点。发现这些知识都了解一些。所以就想着试试自己实现一个简单的RPC框架，即巩固了基础的知识，也能更加深入的了解RPC原理。当然一个完整的RPC框架包含了许多的功能，例如服务的发现与治理，网关等等。本篇只是简单的实现了一个调用的过程。

传参出参分析

一个简单请求可以抽象为两步

那么就根据这两步进行分析，在请求之前我们应该发送给服务端什么信息？而服务端处理完以后应该返回客户端什么信息？

1、在请求之前我们应该发送给服务端什么信息？

由于我们在客户端调用的是服务端提供的接口，所以我们需要将客户端调用的信息传输过去，那么我们可以将要传输的信息分为两类

第一类是服务端可以根据这个信息找到相应的接口实现类和方法
第二类是调用此方法传输的参数信息

那么我们就根据要传输的两类信息进行分析，什么信息能够找到相应的实现类的相应的方法？要找到方法必须要先找到类，这里我们可以简单的用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;
}

2、服务端处理完以后应该返回客户端什么信息？

上面我们分析了客户端应该传输什么信息给服务端，那么服务端处理完以后应该传什么样的返回值呢？这里我们只考虑最简单的情况，客户端请求的线程也会一直在等着，不会有异步处理这一说，所以这么分析的话就简单了，直接将得到的处理结果返回就行了。

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

由于都涉及到了网络传输，所以都要实现序列化的接口

如何获得传参信息并执行？-客户端

上面我们分析了客户端向服务端发送的信息都有哪些？那么我们如何获得这些信息呢？首先我们调用的是接口，所以我们需要写自定义注解然后在程序启动的时候将这些信息加载在Spring容器中。有了这些信息那么我们就需要传输了，调用接口但是实际上执行的确实网络传输的过程，所以我们需要动态代理。那么就可以分为以下两步

初始化信息阶段：将key为接口名，value为动态接口类注册进Spring容器中
执行阶段：通过动态代理，实际执行网络传输

1、初始化信息阶段

由于我们使用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 beanDefinitionSet = provider.findCandidateComponents("com.example.rpcclient.client");
        for (BeanDefinition beanDefinition : beanDefinitionSet){
            if (beanDefinition instanceof AnnotatedBeanDefinition){
                //获得注解上的参数信息
                AnnotatedBeanDefinition annotatedBeanDefinition = (AnnotatedBeanDefinition) beanDefinition;
                String beanClassAllName = beanDefinition.getBeanClassName();
                Map 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注解的代理类

如果对FactoryBean类不了解的可以参见FactoryBean讲解

@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;
    }
}

我们的客户端已经写完了，传给服务端的信息我们也已经拼装完毕了。剩下的工作就简单了，开始编写服务端的代码。

服务端处理完以后应该返回客户端什么信息？-服务端

服务端的代码相比较客户端来说要简单一些。可以简单分为下面三步

拿到接口名以后，通过接口名找到实现类
通过反射进行对应方法的执行
返回执行完的信息

那么我们就根据这三步进行编写代码

1、拿到接口名以后，通过接口名找到实现类

如何通过接口名拿到对应接口的实现类呢？这就需要我们在服务端启动的时候将其对应信息加载进去

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

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

    @Override
    public void run(String... args) throws Exception {
        Map 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存储的就是接口名和其对应的实现类的对应关系。

2、通过反射进行对应方法的执行

此时拿到了对应关系以后就能根据客户端传过来的信息找到相应的实现类中的方法。然后进行执行并返回信息就行

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的类型。

微信关注我们

原文链接：https://blog.51cto.com/u_15191078/2755779

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

面试官：RabbitMQ 如何保障消息 100% 投递成功、消息幂等性？

一、前言我们小伙伴应该都听说够消息中间件MQ，如：RabbitMQ，RocketMQ，Kafka等。引入中间件的好处可以起到抗高并发，削峰，业务解耦的作用。如上图： 1）订单服务投递消息给MQ中间件 2）物流服务监听MQ中间件消息，从而进行消费我们这篇文章讨论一下，如何保障订单服务把消息成功投递给MQ中间件，以RabbitMQ举例。二、分析问题小伙伴们对此会有些疑问，订单服务发起消息服务，返回成功不就成功了吗？如下面的伪代码：上面代码中，一般发送消息就是这么写的，小伙伴们觉得有什么问题吗？下边说一个场景，如果MQ服务器突然宕机了会出现什么情况？是不是我们订单服务发过去的消息全部没有了吗？是的，一般MQ中间件为了提高系统的吞吐量会把消息保存在内存中，如果不作其他处理，MQ服务器一旦宕机，消息将全部丢失。这个是业务不允许的，造成很大的影响，关注公众号码猿技术专栏，获取更多面试资源。三、持久化有经验的小伙伴会说，我知道一个方法就是把消息持久化，RabbitMQ中发消息的时候会有个durable参数可以设置，设置为true，就会持久化。这样的话MQ服务器即使宕机，重启后磁盘...

2021-05-06

429

本文由『Go开源说』第四期 go-zero 直播内容修改整理而成，视频内容较长，拆分成上下篇，本文内容有所删减和重构。大家好，很高兴来到“GO开源说” 跟大家分享开源项目背后的一些故事、设计思想以及使用方法，今天分享的项目是 go-zero，一个集成了各种工程实践的 web 和 rpc 框架。我是Kevin，go-zero 作者，我的 github id 是kevwan。 go-zero 概览 go-zero 虽然是20年8月7号才开源，但是已经经过线上大规模检验了，也是我近20年工程经验的积累，开源后得到社区的积极反馈，在6个多月的时间里，获得了7.8k stars。多次登顶github Go语言日榜、周榜、月榜榜首，并获得了gitee最有价值项目（GVP），开源中国年度最佳人气项目。同时微信社区极为活跃，几千人的社区群，go-zero爱好者们一起交流go-zero使用心得和讨论使用过程中的问题。 go-zero 如何自动管理缓存？缓存设计原理我们对缓存是只删除，不做更新，一旦DB里数据出现修改，我们就会直接删除对应的缓存，而不是去更新。我们看看删除缓存的顺序怎样才是正确的。...

2021-05-06

445

资源下载

更多资源

优质分享App

近一个月的开发和优化，本站点的第一个app全新上线。该app采用极致压缩，本体才4.36MB。系统里面做了大量数据访问、缓存优化。方便用户在手机上查看文章。后续会推出HarmonyOS的适配版本。

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长，特征是大鼻子、头戴帽子、身穿背带裤，还留着胡子。与他的双胞胎兄弟路易基一起，长年担任任天堂的招牌角色。

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题，腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构，目前腾讯云软件源站支持公网访问和内网访问。

Nacos

Nacos /nɑ:kəʊs/ 是 Dynamic Naming and Configuration Service 的首字母简称，一个易于构建 AI Agent 应用的动态服务发现、配置管理和AI智能体管理平台。Nacos 致力于帮助您发现、配置和管理微服务及AI智能体应用。Nacos 提供了一组简单易用的特性集，帮助您快速实现动态服务发现、服务配置、服务元数据、流量管理。Nacos 帮助您更敏捷和容易地构建、交付和管理微服务平台。