《深入理解RPC框架原理与实现 华钟明》使用Netty、Zookeeper等实现一个简单的RPC框架、自定义注解、SPI机制实践与原理分析

前言

读书笔记见往期博客：《深入理解RPC框架原理与实现 华钟明》读书笔记

接下来就是实践练习环节，结合书中最简易的实现的RPC的源码和一些开源的源码，来自己一步一步手敲出一个自己升级版的RPC…通过实践，可以对Netty、Zookeeper等中间件、自定义注解、SPI机制、常见的序列化协议以及网络编程有一定的理解…

一个RPC框架需要具备远程通信方式、通信协议、序列化方式等组件，需要动态代理实现本地存根等。除了基本的远程调用能力，还需要一定的服务治理能力，比如服务的注册发现的注册中心、保护系统流控作用的负载均衡策略、分析错误的链路追踪以及一些基础的日志记录、配置方式等。

将RPC的需求优先级排序，主要分为三步实现：1、实现远程调用相关能力。2、必要的服务治理如注册、发现、负载均衡、路由、容错机制等。3、链路追踪、应用监控、故障告警能力等。因此可以使用下列组件库完成一个简易的RPC框架：Spring、Netty、log4j、Javassit和Cglib、Zookeeper、curator(最为Zookeeper的客户端实现)、Protostuff。

1、升级版RPC原理分析

前面的步骤完成之后，就要考虑整合Spring、SpringBoot使用了，书本中的例子是整合的Spring，通过自定义命名空间，将相关配置写在xml中。为了更加方便，下面进行一个升级、拓展，可以利用注解的原理完成Provider服务声明、Consumer服务引用等。

（1）自定义注解方便整合SpringBoot

根据前面的学习和对RPC本地存根的理解，分析得出需要三个注解来实现PRC功能

• provider 需要@SclRpcService声明接口
• consumer 需要@SclRpcReferencce 本地声明标注到成员属性，引用接口
• provider、consumer都需要@SclRpcScan来扫描上述注解

1、创建需要使用的注解

在simple-rpc-scl模块新建annotation包，创建上述三个注解，注解就是特殊的接口，了解注解的工作原理都知道注解声明之后需要创建代理对象来进行处理完成相应功能，否则起不到任何作用。

package henu.soft.scl.rpc.annotation;

import java.lang.annotation.*;

/**
 * RPC reference annotation, autowire the service implementation class
 *
 */
@Documented
@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.FIELD})
@Inherited
public @interface SclRpcReference {

    /**
     * Service version, default value is empty string
     */
    String version() default "";

    /**
     * Service group, default value is empty string
     */
    String group() default "";

}


package henu.soft.scl.rpc.annotation;

import java.lang.annotation.*;



/**
 * RPC service annotation, marked on the service implementation class
 *
 */
@Documented
@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.TYPE})
@Inherited
public @interface SclRpcService {

    /**
     * Service version, default value is empty string
     */
    String version() default "";

    /**
     * Service group, default value is empty string
     */
    String group() default "";

}

package henu.soft.scl.rpc.annotation;

import henu.soft.scl.rpc.annotation.spring.CustomScannerRegistrar;
import org.springframework.context.annotation.Import;

import java.lang.annotation.*;

/**
 * scan custom annotations
 *
 */



@Target({ElementType.TYPE, ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
@Import(CustomScannerRegistrar.class)
@Documented
public @interface SclRpcScan {

    String[] basePackage();

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79

2、让自己定义的注解发挥作用

2.1、首先需要明白怎么拿到注解定义的属性值的原理

Annontation是Java5开始引入的新特征，中文名称叫注解。它提供了一种安全的类似注释的机制，用来将任何的信息或元数据（metadata）与程序元素（类、方法、成员变量等）进行关联。为程序的元素（类、方法、成员变量）加上更直观更明了的说明，这些说明信息是与程序的业务逻辑无关，并且供指定的工具或框架使用。Annontation像一种修饰符一样，应用于包、类型、构造方法、方法、成员变量、参数及本地变量的声明语句中。注解的本质就是一个继承了 Annotation 接口的接口，只有被解析之后才能发挥作用，而解析一个类或者方法的注解往往有两种形式，一种是编译期直接的扫描，一种是运行期反射。

• 编译器的扫描：指的是编译器在对 java 代码编译字节码的过程中会检测到某个类或者方法被一些注解修饰，这时它就会对于这些注解进行某些处理。典型的就是注解 @Override，一旦编译器检测到某个方法被修饰了 @Override 注解，编译器就会检查当前方法的方法签名是否真正重写了父类的某个方法，也就是比较父类中是否具有一个同样的方法签名。这一种情况只适用于那些编译器已经熟知的注解类，比如 JDK 内置的几个注解，而你自定义的注解，编译器是不知道你这个注解的作用的，当然也不知道该如何处理，往往只是会根据该注解的作用范围来选择是否编译进字节码文件，仅此而已。
• 运行期反射：使用元注解配置一个自定义的注解，本质上是@interface。需要该注解永久存活，以便我们反射获取。虚拟机规范定义了一系列和注解相关的属性表，也就是说，无论是字段、方法或是类本身，如果被注解修饰了，就可以被写进字节码文件。

所以，对于一个类或者接口来说，Class 类中提供了以下一些方法用于反射注解。

• getAnnotation：返回指定的注解，然后可以获取注解中指定的值，其实是代理对象，可以强转
• isAnnotationPresent：判定当前元素是否被指定注解修饰
• getAnnotations：返回所有的注解
• getDeclaredAnnotation：返回本元素的指定注解
• getDeclaredAnnotations：返回本元素的所有注解，不包含父类继承而来的

原理：注解本质是一个继承了Annotation 的特殊接口，其具体实现类是Java 运行时生成的动态代理类。而我们通过反射获取注解时，返回的是Java 运行时生成的动态代理对象$Proxy1。通过代理对象调用自定义注解（接口）的方法，会最终调用AnnotationInvocationHandler 的invoke方法。该方法会从memberValues 这个Map 中索引出对应的值。而memberValues 的来源是Java 常量池。

代理类内部有一个AnnotationInvocatinHandler类型的实例，是Java专门处理注解的handler，主要用来调用被代理接口中的系列方法，

• 如四大方法equals、hashcode、toString、annotationType直接调用的就是handler已经写好的方法
• 被代理接口中自定义注解字节声明的方法，例如我们 Hello 注解的 value 方法。这种情况下，将从我们的注解 memberValues 中获取这个注解属性对应的值。

最后我们再总结一下整个反射注解的工作原理：

1. 首先，我们通过键值对的形式可以为注解属性赋值，像这样：@Hello（value = “hello”）。

2. 接着，你用注解修饰某个元素，编译器将在编译期扫描每个类或者方法上的注解，会做一个基本的检查，你的这个注解是否允许作用在当前位置，最后会将注解信息写入元素的属性表map。

3. 然后，当你进行反射的时候，虚拟机将所有生命周期在 RUNTIME 的注解取出来放到一个 map 类型的memberValues ，并创建一个 AnnotationInvocationHandler 实例，把这个 map 传递给它。

4. 最后，虚拟机将采用 JDK 动态代理机制生成一个目标注解的代理类，并初始化好处理器。

5. 使用的时候，可以把自定义注解标注在Food类字段color上，然后设置value = red。然后可以根据Food的Class对象的getDeclaredFields方法获取全部字段，然后遍历判断isAnnotationPresent，然后getAnnotation获取代理对象实例，接着就可以调用value方法获取red值

那么这样，一个注解的实例就创建出来了，它本质上就是一个代理类，你应当去理解好AnnotationInvocationHandler 中 invoke 方法的实现逻辑，这是核心。一句话概括就是，通过方法名返回注解属性值。

2.2、扫描注册加了自定义注解的类

首先是@SclRpcScan注解功能实现，该注解是扫描指定路径下的包，将标注@SclRpcService扫描出来，然后将对应的类标识、注册到注册中心如Zookeeper，将标注@SclRpcReference扫描出来生成代理对象，在代理对象的invoke方法中调用netty发送请求。

新建CustomScannerRegistrar类，实现ImportBeanDefinitionRegistrar，重写registerBeanDefinitions方法来添加@SclRpcService、@SclRpcReference对应的Scanner，然后调用scan方法进行注册BeanDefinition

/**
 * @author sichaolong
 * @date 2022/7/28 16:56
 */

public class CustomScannerRegistrar implements ImportBeanDefinitionRegistrar, ResourceLoaderAware {
    private static final String SPRING_BEAN_BASE_PACKAGE = "henu.soft";
    private static final String BASE_PACKAGE_ATTRIBUTE_NAME = "basePackage";
    
    private ResourceLoader resourceLoader;
    private static final Logger logger = LoggerFactory.getLogger(CustomScanner.class);

    @Override
    public void setResourceLoader(ResourceLoader resourceLoader) {
        this.resourceLoader = resourceLoader;

    }

    @Override
    public void registerBeanDefinitions(AnnotationMetadata annotationMetadata, BeanDefinitionRegistry beanDefinitionRegistry) {
        //get the attributes and values ​​of RpcScan annotation，获取@SclRpcScan注解设置的basePackage值，也就是扫包的路径
        AnnotationAttributes rpcScanAnnotationAttributes = AnnotationAttributes.fromMap(annotationMetadata.getAnnotationAttributes(SclRpcScan.class.getName()));
        
        String[] rpcScanBasePackages = new String[0];
        // 存在该注解@SclRpcScan
        if (rpcScanAnnotationAttributes != null) {
            // 获取@SclRpcScan 的 basePackage的值
            // get the value of the basePackage property
            rpcScanBasePackages = rpcScanAnnotationAttributes.getStringArray(BASE_PACKAGE_ATTRIBUTE_NAME);
        }
        //未设置 @SclRpcScan 的 basePackage的值
        if (rpcScanBasePackages.length == 0) {
            rpcScanBasePackages = new String[]{((StandardAnnotationMetadata) annotationMetadata).getIntrospectedClass().getPackage().getName()};
        }
        // 添加对应注解的 CustomScanner , CustomScanner是自定义的类实现ClassPathBeanDefinitionScanner，下面介绍
        // Scan the RpcService annotation
        CustomScanner rpcServiceScanner = new CustomScanner(beanDefinitionRegistry, SclRpcService.class);
        // Scan the Component annotation
        CustomScanner springBeanScanner = new CustomScanner(beanDefinitionRegistry, Component.class);
       
        if (resourceLoader != null) {
            rpcServiceScanner.setResourceLoader(resourceLoader);
            springBeanScanner.setResourceLoader(resourceLoader);
        }
        // 具体扫描
        int springBeanAmount = springBeanScanner.scan(SPRING_BEAN_BASE_PACKAGE);
        logger.info("springBeanScanner扫描的数量 [{}]", springBeanAmount);
        int rpcServiceCount = rpcServiceScanner.scan(rpcScanBasePackages);
        logger.info("rpcServiceScanner扫描的数量 [{}]", rpcServiceCount);

    }

}


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

然后就是到了自定义类CustomScanner的scan方法，scan方法内部也没干什么事情，内部还是委托Spring的ClassPathBeanDefinitionScanner.doScan()方法来扫描，然后在注册到IoC容器，注册到容器之后，就可以利用AOP在初始化前后调用postProcessBeforeInitialization、postProcessAfterInitialization方法完成相关增强操作

/**
 * @author sichaolong
 * @date 2022/7/28 16:55
 */
public class CustomScanner extends ClassPathBeanDefinitionScanner {

    public CustomScanner(BeanDefinitionRegistry registry, Class<? extends Annotation> annoType) {
        super(registry);
        super.addIncludeFilter(new AnnotationTypeFilter(annoType));
    }

    @Override
    public int scan(String... basePackages) {
        // 委托Spring的ClassPathBeanDefinitionScanner来扫描注册BeanDefinition，注册进Spring容器
        return super.scan(basePackages);
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

注册进BeanDefinition之后，就可以通过AOP来拦截，新建SpringBeanPostProcessor实现BeanPostProcessor类，重写Bean初始化前置、后置增强，也就是

• 在前置增强postProcessBeforeInitialization这里将Provider由@SclRpcService标注的服务注册到注册中心。
• 在后置增强postProcessBeforeInitialization这里将Consumer由@SclRpcReference标注的本地存根相关的接口Bean保存到生成的代理对象中，代理对象也就是通过反射按照本地引用接口生成，然后保存了

package henu.soft.scl.rpc.annotation.spring;


/**
 * @author sichaolong
 * @date 2022/7/28 16:56
 */

@Component
public class SpringBeanPostProcessor implements BeanPostProcessor {

    private static final Logger logger = LoggerFactory.getLogger(SpringBeanPostProcessor.class);
	
    // provider 服务注册，也就是 @SclRpcService 标注服务 注册到 配置的注册中心
    private final ServiceProvider serviceProvider;
    // consumer 客户端远程调用的SPI
    private final RpcRequestTransport rpcClient;

    public SpringBeanPostProcessor() {
        this.serviceProvider = SingletonFactory.getInstance(ZkServiceProviderImpl.class);
        this.rpcClient = ExtensionLoader.getExtensionLoader(RpcRequestTransport.class).getExtension("netty");
    }

    @SneakyThrows
    @Override
    public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
        // 该BeanDefinition上有 @SclRpcService中 注解
        if (bean.getClass().isAnnotationPresent(SclRpcService.class)) {
            logger.info("[{}] is annotated with  [{}]", bean.getClass().getName(), SclRpcService.class.getCanonicalName());
            // get RpcService annotation
            SclRpcService rpcService = bean.getClass().getAnnotation(SclRpcService.class);
            // build RpcServiceProperties
            RpcServiceConfig rpcServiceConfig = RpcServiceConfig.builder()
                    .group(rpcService.group())
                    .version(rpcService.version())
                    .service(bean).build();
            // provider 服务注册到注册中心
            serviceProvider.publishService(rpcServiceConfig);
        }
        return bean;
    }

    @Override
    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
        Class<?> targetClass = bean.getClass();
        // 获取 BeanDefinition 的属性字段
        Field[] declaredFields = targetClass.getDeclaredFields();
        for (Field declaredField : declaredFields) {
            // 遍历成员属性字段，判断是否标注 @SclRpcReference注解
            SclRpcReference rpcReference = declaredField.getAnnotation(SclRpcReference.class);
            if (rpcReference != null) {
                RpcServiceConfig rpcServiceConfig = RpcServiceConfig.builder()
                        .group(rpcReference.group())
                        .version(rpcReference.version()).build();
                // 创建 RpcClientProxy 代理对象生成类，然后getProxy方法获得代理对象使用Netty发送请求，内部成员变量RpcRequestTransport就是调用netty发请求
                RpcClientProxy rpcClientProxy = new RpcClientProxy(rpcClient, rpcServiceConfig);
                Object clientProxy = rpcClientProxy.getProxy(declaredField.getType());
                declaredField.setAccessible(true);
                try {
                    // 将 标注@SclRpcReference注解 的Bean 设置到 clientProxy代理对象的字段中，方便等会调用invoke方法调用netty发请求
                    declaredField.set(bean, clientProxy);
                } catch (IllegalAccessException e) {
                    e.printStackTrace();
                }
            }

        }
        return bean;
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71

标注@SclRpcRefernce注解的成员变量（接口）生成的代理对象rpcClientProxy，在代理对象的invoke方法调用netty发送接受请求。

package henu.soft.scl.rpc.proxy;



/**
 * @author sichaolong
 * @date 2022/7/29 11:04
 */
@Slf4j
public class RpcClientProxy implements InvocationHandler {

    private static final String INTERFACE_NAME = "interfaceName";

    /**
     * Used to send requests to the server.And there are two implementations: socket and netty
     */
    // 调用netty发请求的类
    private final RpcRequestTransport rpcRequestTransport;
    // 发送请求携带的相关信息
    private final RpcServiceConfig rpcServiceConfig;

    public RpcClientProxy(RpcRequestTransport rpcRequestTransport, RpcServiceConfig rpcServiceConfig) {
        this.rpcRequestTransport = rpcRequestTransport;
        this.rpcServiceConfig = rpcServiceConfig;
    }


    public RpcClientProxy(RpcRequestTransport rpcRequestTransport) {
        this.rpcRequestTransport = rpcRequestTransport;
        this.rpcServiceConfig = new RpcServiceConfig();
    }
	// 反射获取代理对象
    /**
     * get the proxy object
     */
    @SuppressWarnings("unchecked")
    public <T> T getProxy(Class<T> clazz) {
        return (T) Proxy.newProxyInstance(clazz.getClassLoader(), new Class<?>[]{clazz}, this);
    }

    /**
     * This method is actually called when you use a proxy object to call a method.
     * The proxy object is the object you get through the getProxy method.
     */
    @SneakyThrows
    @SuppressWarnings("unchecked")
    @Override
    public Object invoke(Object proxy, Method method, Object[] args) {
        log.info("invoked method: [{}]", method.getName());
        RpcRequest rpcRequest = RpcRequest.builder().methodName(method.getName())
                .parameters(args)
                .interfaceName(method.getDeclaringClass().getName())
                .paramTypes(method.getParameterTypes())
                .requestId(UUID.randomUUID().toString())
                .group(rpcServiceConfig.getGroup())
                .version(rpcServiceConfig.getVersion())
                .build();
        RpcResponse<Object> rpcResponse = null;
        if (rpcRequestTransport instanceof NettyRpcClient) {
            // netty 异步发请求
            CompletableFuture<RpcResponse<Object>> completableFuture = (CompletableFuture<RpcResponse<Object>>) rpcRequestTransport.sendRpcRequest(rpcRequest);
            rpcResponse = completableFuture.get();
        }
        	// 普通socket 发请求
//        if (rpcRequestTransport instanceof SocketRpcClient) {
//            rpcResponse = (RpcResponse