登录
    Technology changes quickly but people's minds change slowly.

自己动手实现简单的微服务rpc调用

技术宅 破玉 1691次浏览 0个评论

什么是rpc(来自百度百科)

RPC（Remote Procedure Call）—远程过程调用，它是一种通过网络从远程计算机程序上请求服务，而不需要了解底层网络技术的协议。RPC协议假定某些传输协议的存在，如TCP或UDP，为通信程序之间携带信息数据。在OSI网络通信模型中，RPC跨越了传输层和应用层。RPC使得开发包括网络分布式多程序在内的应用程序更加容易。

RPC采用客户机/服务器模式。请求程序就是一个客户机，而服务提供程序就是一个服务器。首先，客户机调用进程发送一个有进程参数的调用信息到服务进程，然后等待应答信息。在服务器端，进程保持睡眠状态直到调用信息到达为止。当一个调用信息到达，服务器获得进程参数，计算结果，发送答复信息，然后等待下一个调用信息，最后，客户端调用进程接收答复信息，获得进程结果，然后调用执行继续进行。

参考资料：rpc原理
参考代码：rpc代码

Rpc要解决的问题

RPC要解决的两个问题：

  1. 解决分布式系统中，服务之间的调用问题。
  2. 远程调用时，要能够像本地调用一样方便，让调用者感知不到远程调用的逻辑。

实际情况下，RPC很少用到http协议来进行数据传输，毕竟我只是想传输一下数据而已，何必动用到一个文本传输的应用层协议呢，我为什么不直接使用二进制传输？比如直接用Java的Socket协议进行传输？

要实现一个RPC不算难，难的是实现一个高性能高可靠的RPC框架。

比如，既然是分布式了，那么一个服务可能有多个实例，你在调用时，要如何获取这些实例的地址呢？

这时候就需要一个服务注册中心，比如在Dubbo里头，就可以使用Zookeeper作为注册中心，在调用时，从Zookeeper获取服务的实例列表，再从中选择一个进行调用。

那么选哪个调用好呢？这时候就需要负载均衡了，于是你又得考虑如何实现复杂均衡，比如Dubbo就提供了好几种负载均衡策略。

这还没完，总不能每次调用时都去注册中心查询实例列表吧，这样效率多低呀，于是又有了缓存，有了缓存，就要考虑缓存的更新问题，blablabla……

你以为就这样结束了，没呢，还有这些：

  • 客户端总不能每次调用完都干等着服务端返回数据吧，于是就要支持异步调用；
  • 服务端的接口修改了，老的接口还有人在用，怎么办？总不能让他们都改了吧？这就需要版本控制了；
  • 服务端总不能每次接到请求都马上启动一个线程去处理吧？于是就需要线程池；
  • 服务端关闭时，还没处理完的请求怎么办？是直接结束呢，还是等全部请求处理完再关闭呢？
  • ……

如此种种，都是一个优秀的RPC框架需要考虑的问题。

我们当前的目标是先实现个简单的rpc 调用，来理解我们分布式系统中服务调用的基本原理。

简单Rpc实现

实现的话，当然就分为客户端和服务端两步，参照我们之前介绍的 feignclient 我们来完成一个简单的rpcclient

整个流程的话，就是客户端调用远程服务端的某个方法，服务端最终返回数据给客户端。
<

客户端实现

客户端要发送什么数据给服务端呢？

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

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

所以我们首先提供实体来保存这两类数据，由于是网络传输，还要保证我们的实体要序列化。

请求的话，我们就用RpcRequest 来保存

/**
 * rpc  request
 */
@Data
public class RpcRequest implements Serializable {

    private static final long serialVersionUID = -8558250604858740384L;
    private String className;
    private String methodName;
    private Class [] parameTypes;
    private Object [] parameters;
}

响应的话，我们用RpcResponse 来保存

/**
 * rpc response
 */
@Data
public class RpcResponse implements Serializable {
    private static final long serialVersionUID = 9201838302447943808L;
    private Object result;
}

那么我们如何将用户的请求封装到客户端，客户端再发给服务端呢？

借助于我们之前理解的 FeignClient和spring 容器的知识，我们可以利用注解和动态代理来实现请求封装和服务调用。

我们提供一个简单的注解RpcClient：


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

}

上面注解的作用和FeignClient 作用一样，来标示这是一个远程调用的接口，以便spring能够扫描到，并注入容器中。此过程，我们需要提供代理类，以便动态代理能够调用，提供工厂类，生成代理类，提供注册类，将bean注入到spring 容器。spring 已经为我们提供了这方面的开发流程，详细的问题，请自行学习spring的相关文章。

首先提供代理类，代理类是服务调用的核心，就是通过执行代理类的 invoke方法，将用户信息以及相关类信息序列化，通过 Socket 发给服务端，并解析服务端响应的数据，最终返回给用户。

@Slf4j
@Component
public class RpcDynamicProxy 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);
        RpcResponse response = new RpcResponse();
        Gson gson =new Gson();
        try{
            //从服务器端接收数据有个时间限制（系统自设，也可以自己设置），超过了这个时间，便会抛出该异常
            String responsJson = buf.readLine();
            response = gson.fromJson(responsJson, RpcResponse.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){
        RpcRequest request = new RpcRequest();
        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;
    }
}

提供 工厂类，生成代理类的对象

public class RpcClientFactoryBean implements FactoryBean {

    private Class classType;
    @Autowired
    private RpcDynamicProxy rpcDynamicProxy;

    public RpcClientFactoryBean(Class classType) {
        this.classType = classType;
    }

    @Override
    public Object getObject() throws Exception {
        ClassLoader classLoader = classType.getClassLoader();
        Object object = Proxy.newProxyInstance(classLoader,new Class[]{classType},rpcDynamicProxy);
        return object;
    }

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

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

提供注册类，扫描相关RpcClient 注解的类，将bean注入spring 容器

public class RpcClientsRegistrar implements ImportBeanDefinitionRegistrar {
    @Override
    public void registerBeanDefinitions(AnnotationMetadata importingClassMetadata, BeanDefinitionRegistry registry) {
        registerRpcClients(importingClassMetadata,registry);
    }

    /**
     * 根据注解 注册 rpcClients
     * @param metadata
     * @param registry
     */
    public void registerRpcClients(AnnotationMetadata metadata,
                                     BeanDefinitionRegistry registry) {
        ClassPathScanningCandidateComponentProvider provider = getScanner();
        //设置扫描器
        provider.addIncludeFilter(new AnnotationTypeFilter(RpcClient.class));
        //扫描此包下的所有带有@RpcClient的注解的类
        Set beanDefinitionSet = provider.findCandidateComponents("com.magicdu.rpc.rpcclient");
        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(RpcClientFactoryBean.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() &amp;&amp; beanDefinition.getMetadata().isIndependent();
            }
        };
    }
}

然后将 注册类，引入 spring 配置，以便启动时扫描

@Configuration
@Import(RpcClientsRegistrar.class)
public class RpcConfig {
}

这样客户端我们就写完了。

服务端代码

服务端简单来说，就是开个端口，接受客户端发送的参数，找到对应的方法，调用完，返回数据就完了。

我们将所有的客户端调用的方法都用 @Service 注解标记，服务端启动时我们将所有Service注解的方法缓存起来，客户端调用时，通过反射，我们便能执行服务端的方法了。并且扫描完就启动相关Socket，我们提供一个 Socket ，等待客户端调用。

@Component
@Slf4j
public class RpcServiceConfig 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());
            }
        }
        startPort();
    }
    private String getClassName(String beanClassName){
        String className = beanClassName.substring(beanClassName.lastIndexOf(".")+1);
        className = className.substring(0,1).toLowerCase() + className.substring(1);
        return className;
    }

    public void  startPort() throws IOException {
                //服务端在20006端口监听客户端请求的TCP连接
        ServerSocket server = new ServerSocket(20006);
        Socket client = null;
        boolean f = true;
        while (f) {
            //等待客户端的连接，如果没有获取连接
            client = server.accept();
           System.out.println("与客户端连接成功！");
           //为每个客户端连接开启一个线程
            new Thread(new ServerThread(client)).start();
        }
        server.close();
    }
}

提供线程，供客户端调用，如果收到客户端请求，就去执行对应的方法。

@Slf4j
public class ServerThread implements Runnable {

    private Socket client = null;

    public ServerThread(Socket client) {
        this.client = client;
    }

    @Override
    public void run() {
        try {
            //获取Socket的输出流，用来向客户端发送数据
            PrintStream out = new PrintStream(client.getOutputStream());
            //获取Socket的输入流，用来接收从客户端发送过来的数据
            BufferedReader buf = new BufferedReader(new InputStreamReader(client.getInputStream()));
            boolean flag = true;
            while (flag) {
                //接收从客户端发送过来的数据
                String str = buf.readLine();
                if (str == null || "".equals(str)) {
                    flag = false;
                } else {
                    out.println(CommonDeal.getInvokeMethodMes(str));
                }
            }
            out.close();
            client.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


}

执行方法的实现如下：

@Slf4j
public class CommonDeal {

    public static String getInvokeMethodMes(String str){
        GsonBuilder gsonBuilder = new GsonBuilder();
        gsonBuilder.registerTypeAdapterFactory(new ClassTypeAdapterFactory());
        Gson gson = gsonBuilder.create();
        RpcRequest request = gson.fromJson(str, RpcRequest.class);
        return gson.toJson(invokeMethod(request));
    }

    private static RpcResponse invokeMethod(RpcRequest request) {
        String className = request.getClassName();
        String methodName = request.getMethodName();
        Object[] parameters = request.getParameters();
        Class[] parameTypes = request.getParameTypes();
        Object o = RpcServiceConfig.rpcServiceMap.get(className);
        RpcResponse response = new RpcResponse();
        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的思路以及简单的实现。

华裳绕指柔, 版权所有丨如未注明 , 均为原创|转载请注明自己动手实现简单的微服务rpc调用！
喜欢 (0)
发表我的评论
取消评论
表情 贴图 加粗 删除线 居中 斜体 签到

Hi，您需要填写昵称和邮箱！

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址