RPC框架和简易RPC实现

项目背景

中途加入一个团队，团队实现一个私有化的分布式服务，因为是私有化，不用大规模集群，服务已经有个简易RPC，
实现思路大致是：
1 暴露一个服务需要自己定义一个服务handler
2 消费者通过http发送输入输出函数名称到服务提供者
3 http获取结果反序列化获得服务输出

概括来说就是http强行rpc，而且复杂的输入输出配置还不如直接用http来实现。“重构”是看到这个实现的第一反应。

重构之前准备

1 了解一下为什么之前rpc协议写成这个样子，大概得到几个观点：
* 之前是demo期，快速实现需求
* 商户不希望开多个端口，所有尽量复用http服务的端口
* 在运行中注入提供方的url，进行多个不同服务的直连
* 私有化小规模部署，不想用复杂的注册中心
* 私有化小规模部署，简化安装依赖
2 团队情况
* Springboot，java和python语言
* 熟练Dubbo，了解grpc

主流RPC框架

thrift
ZeroC Ice
gRPC
Dubbo

thrift

参考：https://blog.csdn.net/lupengfei1009/article/details/100934794

Facebook研发的轻量级RPC，可以配合服务器一起运行。
支持多语言
二进制格式传输数据，比XML和JSON体积小，适合高并发，大数据量，多语言的环境。
C/S结构

不选择的原因：
1 需要额外的rpc端口
2 c / s结构，明确的服务端和客户端，但是实际上，rpc更多时候，一个服务既是客户端又是服务端。
3 springboot集成有一个依赖包，mac下开发可能需要安装thrift才能使用代码生成器（这个点不是核心考虑点）
4 团队不太熟悉，但是学习成本很低，也不是核心考虑点

Dubbo

Dubbo的强大行业内已经众所周知了，其原理图如下：

image.png

但是：
1 直连模式（无注册中心）在注解中配置（单个url），无法在运行中动态指定多个不同节点。
2 直连模式调试Reference的时候遇到一些bug（中途有一些特殊的空指针异常），提了issue建议升级版本，没有提供者的时候，注入Reference为空，启动服务提供者之后，Reference不会重新注入（不知道新版是否解决）。
3 也需要开放特定的Dubbo端口

简易RPC框架实现

为了迎合项目特定的需求，考虑简化实现一个RPC，几乎没有多余的第三方依赖，使用方式类似Dubbo，实现方式如下：
1 通过 @RPCService 注入服务提供者，直接利用Springboot的IOC注入
2 通过 @RPCReference 注入服务消费者，服务通过动态代理，调用远程服务
3 通过setNode在运行中配置直连节点信息，比如节点端口，ip，httpPath，host等。（因为http直接通信，复用了应用容器的http端口，不需要新的端口）

代码结构

├── core
│ ├── config
│ │ ├── ConsumerConfig.java 消费者配置
│ │ ├── ProviderConfig.java 提供者配置
│ │ └── annotation
│ │ ├── RPCReference.java 消费者注解
│ │ └── RPCService.java 提供者注解
│ ├── context
│ │ ├── ConsumerContext.java 消费者上下文
│ │ └── ProviderContext.java 提供者上下文
│ ├── domain
│ │ ├── Node.java 节点信息，用于配置特定的ip，端口，token等
│ │ ├── RPCRequest.java 消费者发送的数据结构
│ │ ├── RPCResponse.java 消费者从提供者那得到的返回结果
│ │ ├── RPCSerializer.java 序列化协议实现
│ │ └── interfaces
│ │ ├── Formater.java
│ │ └── Parser.java
│ ├── exception
│ │ └── RPCException.java 自定义异常
│ └── proxy
│ ├── ConsumerProxyFactory.java （rpc核心逻辑）消费者调用放的动态代理，是实际执行的方法
│ └── ProviderProxyFactory.java （rpc核心逻辑）注入提供者服务，以及服务通过反射实际执行
├── spring
│ └── factory
│ ├── RPCClassPathBeanDefinitionScanner.java 利用IOC注入服务
│ ├── RPCReferenceAnnotationBeanPostProcessor.java 利用IOC注入服务并通过反射替换Reference的实际方法
│ └── RPCServiceAnnotationBeanPostProcessor.java 利用IOC注入服务
└── utils
├── HttpClientUtil.java 通信用的Http工具类
└── HttpUtil.java

核心代码 @RPCServide服务的注入

@Component
@Slf4j
public class RPCServiceAnnotationBeanPostProcessor implements BeanFactoryPostProcessor, ApplicationContextAware {


    private final Logger logger = LoggerFactory.getLogger(getClass());

    private ApplicationContext applicationContext;

    public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {
        BeanDefinitionRegistry registry = (BeanDefinitionRegistry) beanFactory;
        RPCClassPathBeanDefinitionScanner scanner = new RPCClassPathBeanDefinitionScanner(registry);

        scanner.setResourceLoader(this.applicationContext);

        BeanNameGenerator beanNameGenerator = resolveBeanNameGenerator(registry);
        scanner.setBeanNameGenerator(beanNameGenerator);

        scanner.addIncludeFilter(new AnnotationTypeFilter(RPCService.class));

        scanner.scan("");
    }


    /**
     * It'd better to use BeanNameGenerator instance that should reference
     * {@link ConfigurationClassPostProcessor},
     * thus it maybe a potential problem on bean name generation.
     *
     * @param registry {@link BeanDefinitionRegistry}
     * @return {@link BeanNameGenerator} instance
     * @see SingletonBeanRegistry
     * @see AnnotationConfigUtils#CONFIGURATION_BEAN_NAME_GENERATOR
     * @see ConfigurationClassPostProcessor#processConfigBeanDefinitions
     * @since 2.5.8
     */
    private BeanNameGenerator resolveBeanNameGenerator(BeanDefinitionRegistry registry) {

        BeanNameGenerator beanNameGenerator = null;

        if (registry instanceof SingletonBeanRegistry) {
            SingletonBeanRegistry singletonBeanRegistry = SingletonBeanRegistry.class.cast(registry);
            beanNameGenerator = (BeanNameGenerator) singletonBeanRegistry.getSingleton(CONFIGURATION_BEAN_NAME_GENERATOR);
        }

        if (beanNameGenerator == null) {

            if (logger.isInfoEnabled()) {

                logger.info("BeanNameGenerator bean can't be found in BeanFactory with name ["
                        + CONFIGURATION_BEAN_NAME_GENERATOR + "]");
                logger.info("BeanNameGenerator will be a instance of " +
                        AnnotationBeanNameGenerator.class.getName() +
                        " , it maybe a potential problem on bean name generation.");
            }
            beanNameGenerator = new AnnotationBeanNameGenerator();

        }

        return beanNameGenerator;
    }


    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }
}

核心代码 @RPCReferenceServide服务的注入

注入过程类似，不过增加了一个注解下方法的判断
如果被RPCReferenceService注解，动态代理到新的方法上
核心代码如下：

private void setField(Object bean, Field field) {
        if (!field.isAccessible()) {
            field.setAccessible(true);
        }
        try {
            // field.set(bean, applicationContext.getBean(field.getType()));
            if (field.getType().isInterface()) {
                field.set(bean, applicationContext.getBean(ConsumerProxyFactory.class).create(field.getType()));
            } else {
                field.set(bean, applicationContext.getBean(ConsumerProxyFactory.class).create(field.getType().getInterfaces()[0]));
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        log.info("set field :", field.getName());
    }

这里既然用到了动态代理，补充一下，动态代理主要有两种方式：

JDK动态代理，jre提供给我们的类库，代理接口
cglib动态代理，通过“继承”可以继承父类所有的公开方法，然后可以重写这些方法，在重写时对这些方法增强，这就是cglib的思想。

此处用的是JDK动态代理，不依赖第三方安装包，但是如果要追求灵活性的话，cglib其实是更好的选择。其中动态代理的核心逻辑如下：

@Slf4j
public class ConsumerProxyFactory implements InvocationHandler {

    private ConcurrentHashMap<String, ConsumerConfig> consumerConfigConcurrentHashMap = new ConcurrentHashMap<>();

    /* create()创建工厂bean speakInterface Class<?> interfaceClass = Class.forName(clazz); */
    public Object create(Class<?> interfaceClass) {
        return Proxy.newProxyInstance(interfaceClass.getClassLoader(), new Class[]{interfaceClass}, this);
    }

    /**
     * 实现InvocationHandler的接口<br>
     * 遇到的问题,Spring容器可能访问被代理类的的实例的toString()方法
     */
    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
     // 这里通过反射的方式写实际的方法执行逻辑
     // 发送实际的HTTP请求，发送：类名，方法名，参数名，参数
    }
}

提供方http获取string反序列化之后，获取容器中注入的方法

    public Object getBeanByClass(Class<?> clazz) throws RPCException {
        Object bean = providers.get(clazz);
        if (bean != null) {
            return bean;
        }
        throw new RPCException("xxx", clazz);
    }

然后通过反射执行返回结果http返回给消费者。

消费方定义一个controller来接收请求，作为通信桥梁

@Autowired
  private ProviderProxyFactory providerProxyFactory;

  @RequestMapping(value = "/rpc", method = RequestMethod.POST)
  public String rpc(String resquestString) {
    String result = providerProxyFactory.handleHttpContent(resquestString);
    return result;
  }

使用方式举例：

1 接口IUser
2 通过RPCService注入服务UserService作为提供者

@RPCService
public class UserService implements IUser{
  public String getName(String id) {
    return "name:" + id;
  }
 }
}

3 注入ip和port，调用本地服务和远程服务

@RequestMapping("/rpc")
@RestController
public class UserController {

  @Autowired
  private UserService userService; // Springboot的注入方式，本地服务

  @RPCReference
  private IUser userService2; // 注入远程服务

  @RequestMapping(value = "/getUserName", method = RequestMethod.POST)
  public JSONObject add(@RequestBody User user) {
    String a = userService.getName("1");  // 输出“name:1”

    Node node = new Node();
    node.setIp("172.16.105.142");
    node.setPort("8088");
    userService2.setNode(node);
    String b = userService2.getName("2");  // 远程服务返回"name:2"

    JSONObject result = new JSONObject();
    result.put("local", a);
    result.put("remote", b);
    return result;
  }
}

改进

1 有的项目结构是

root_path
- package1
- package2
  - WebApplication.java
    在WebApplication中引用package1中内容，通常Springboot通过@ComponentScan(root_path)来指定扫描跟路径。

所以在rpc注入provider的时候，也提供一个扫描根路径的注解配置（也是仿的Dubbo）

在注解中使用import

@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Import(SimpleRPCComponentScanRegistrar.class) // 这里
public @interface SimpleRPCComponentScan {

    String[] value() default {};

    String[] basePackages() default {};
}

import的SimpleRPCComponentScanRegistrar继承ImportBeanDefinitionRegistrar，并覆写registerBeanDefinitions方法

public class SimpleRPCComponentScanRegistrar implements ImportBeanDefinitionRegistrar {

    @Override
    public void registerBeanDefinitions(AnnotationMetadata importingClassMetadata, BeanDefinitionRegistry registry) {
        Set<String> packagesToScan = getPackagesToScan(importingClassMetadata);

        registerServiceAnnotationBeanPostProcessor(packagesToScan, registry);
    }
}

使用方式

@SpringBootApplication
@SimpleRPCComponentScan("xxx") // 这里
public class DemoApplication {

    public static void main(String[] args) {
        SpringApplication.run(DemoApplication.class, args);
    }

}

2 在注入提供者的时候注入的类可能被代理过，导致获取的接口名字不准确，解决方式，使用AopUtils.getTargetClass(obj) 获取真正的类

RPC框架和简易RPC实现