Netty进阶实现自定义Rpc

项目地址：xz125/Rpc-msf (github.com)

1 项目架构：

RPC 框架包含三个最重要的组件，分别是客户端、服务端和注册中心。在一次 RPC 调用流程中，这三个组件是这样交互的：

服务端(provider)在启动后会将它提供的服务列表和地址信息发布到注册中心。

客户端(client)会通过本地代理模块 Proxy 发现注册中心的服务信息，客户端从服务列表中选取其中一个的服务地址，并将数据通过网络发送给服务端。

服务端接收到数据后进行解码，得到请求信息；

服务端根据解码后的请求信息调用对应的服务，然后将调用结果返回给客户端。

2 项目依赖

2.1 使用maven聚合工程

rpc 父工程

consumer，服务消费者，是rpc的子工程，依赖于rpc-client-spring-boot-starter。

provider，服务提供者，是rpc的子工程，依赖于rpc-server-spring-boot-starter。

provider-api，服务提供者暴露的服务API，是rpc的子工程。

rpc-client-spring-boot-starter，rpc客户端starter，封装客户端发起的请求过程（动态代理、网络通信）。

rpc-core，RPC核心依赖，负载均衡策略、消息协议、协议编解码、序列化、请求响应实体、服务注册发现。

rpc-server-spring-boot-starter，rpc服务端starter，负责发布 RPC 服务，接收和处理 RPC 请求（动态代理、网络通信），反射调用服务端。

项目依赖图

3 项目解析

3.1发布服务和消费服务

对于发布的服务需要使用 @RpcService 注解标识，复合注解，基于 @Service

// !!!注解定义
@Target(ElementType.TYPE)
@Documented
@Retention(RetentionPolicy.RUNTIME)
@Service
public @interface RpcService {
    /**
     *  暴露服务接口类型
     * @return
     */
    Class<?> interfaceType() default Object.class;

    /**
     *  服务版本
     * @return
     */
    String version() default "1.0";
}

// !!!实现类定义
@Slf4j
@RpcService(interfaceType = HelloWorldService.class, version = "1.0")
public class HelloWorldServiceImpl implements HelloWorldService {
    @Override
    public String sayHello(String name) {
        log.info("您好：" + name + ", rpc 调用成功");
        return String.format("您好：%s, rpc 调用成功", name);
    }
}

消费服务需要使用 @RpcAutowired 注解标识，复合注解，基于 @Autowired

// 注解定义
@Target({ElementType.CONSTRUCTOR, ElementType.METHOD, ElementType.PARAMETER, ElementType.FIELD, ElementType.ANNOTATION_TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Autowired
public @interface RpcAutowired {

    String version() default "1.0";
}
 
// controller 实现。
@Controller
public class HelloWorldController {

    @RpcAutowired
    private HelloWorldService helloWorldService;

    @GetMapping("/hello/world")
    public ResponseEntity<String> pullServiceInfo(@RequestParam("name") String name){
        return  ResponseEntity.ok(helloWorldService.sayHello(name));
    }
}

3.2 动态代理

基于jdk接口的动态代理，客户端不能切换（rpc-client-spring-boot-starter模块 proxy 包）

原理是服务消费者启动的时候有个 RpcClientProcessor bean 的后置处理器，会扫描ioc容器中的bean,如果这个bean有属性被@RpcAutowired修饰，就给属性动态赋代理对象。

3.3 服务注册发现

本项目使用ZK做的，实现在 `rpc-core` 模块 com.msf.core.discovery 包下面是服务发现， com.msf.core.register 包下面是服务注册。服务提供者启动后，RpcServerProvider 会获取到被 @RpcService 修饰的bean，将服务元数据注册到zk上。

3.4 负载均衡策略

负载均衡定义在rpc-core中，目前支持轮询（FullRoundBalance）和随机（RandomBalance），默认使用随机策略。由rpc-client-spring-boot-starter指定，可以通过配置文件进行修改。

@Primary
 @Bean(name = "loadBalance")
 @ConditionalOnMissingBean
 @ConditionalOnProperty(prefix = "rpc.client", name = "balance", havingValue = "randomBalance", matchIfMissing = true)
 public LoadBalance randomBalance() {
     return new RandomBalance();
 }

 @Bean(name = "loadBalance")
 @ConditionalOnMissingBean
 @ConditionalOnProperty(prefix = "rpc.client", name = "balance", havingValue = "fullRoundBalance")
 public LoadBalance loadBalance() {
     return new FullRoundBalance();
 }

可以在消费者中配置 rpc.client.balance=fullRoundBalance 替换，也可以自己定义，通过实现接口 LoadBalance，并将创建的类加入IOC容器即可。

3.5自定义消息协议、编解码。

所谓协议，就是通信双方事先商量好规则，服务端知道发送过来的数据将如何解析。

+---------------------------------------------------------------+

+---------------------------------------------------------------+

| 状态 1byte | 消息 ID 8byte | 数据长度 4byte |

+---------------------------------------------------------------+

| 数据内容（长度不定） |

+---------------------------------------------------------------+

魔数：魔数是通信双方协商的一个暗号，通常采用固定的几个字节表示。魔数的作用是防止任何人随便向服务器的端口上发送数据。例如 java Class 文件开头就存储了魔数 0xCAFEBABE，在加载 Class 文件时首先会验证魔数的正确性

协议版本号：随着业务需求的变化，协议可能需要对结构或字段进行改动，不同版本的协议对应的解析方法也是不同的。

序列化算法：序列化算法字段表示数据发送方应该采用何种方法将请求的对象转化为二进制，以及如何再将二进制转化为对象，如 JSON、Hessian、Java 自带序列化等。

报文类型：在不同的业务场景中，报文可能存在不同的类型。RPC 框架中有请求、响应、心跳等类型的报文。（可以是请求报文，或者是响应报文等等）

状态：状态字段用于标识请求是否正常（SUCCESS、FAIL）。

消息ID：请求唯一ID，通过这个请求ID将响应关联起来，也可以通过请求ID做链路追踪。

数据长度：标明数据的长度，用于判断是否是一个完整的数据包

数据内容：请求体内容

3.6 编解码

编解码实现在 rpc-core 模块，在包 com.msf.core.codec下。

3.6.1 如何实现编解码？

编码利用 netty 的 MessageToByteEncoder 类实现。实现 encode 方法，MessageToByteEncoder 继承 ChannelOutboundHandlerAdapter 。编码就是将请求数据写入到 ByteBuf 中。

解码是利用 netty 的 ByteToMessageDecoder 类实现。实现 decode 方法，ByteToMessageDecoder 继承 ChannelInboundHandlerAdapter。解码就是将 ByteBuf 中数据解析出请求的数据。解码要注意 TCP 粘包和拆包问题。

本项目如何解决的？

使用的是消息长度 + 消息内容的形式。在解码器 RpcDecoder 中读取固定长度数据。

3.7 序列化与反序列化

序列化和反序列化在 rpc-core 模块 com.msf.core.serialization 包下，提供了 HessianSerialization 和 JsonSerialization 序列化。默认使用 HessianSerialization 序列化。用户也可以根据项目中实现接口进行自定义。

public static SerializationTypeEnum parseByName(String typeName) {
        for (SerializationTypeEnum typeEnum : SerializationTypeEnum.values()) {
            if (typeEnum.name().equalsIgnoreCase(typeName)) {
                return typeEnum;
            }
        }
        return HESSIAN;
    }

    public static SerializationTypeEnum parseByType(byte type) {
        for (SerializationTypeEnum typeEnum : SerializationTypeEnum.values()) {
            if (typeEnum.getType() == type) {
                return typeEnum;
            }
        }
        return HESSIAN;
    }

3.8 网络传输

使用netty代码，值得注意的是 handler 的顺序不能弄错，编码是出站操作（可以放在入站后面），解码和收到响应都是入站操作，解码要在前面。

bootstrap.group(eventLoopGroup).channel(NioSocketChannel.class)
.handler(new ChannelInitializer<SocketChannel>() {
    @Override
    protected void initChannel(SocketChannel socketChannel) throws Exception {
        socketChannel.pipeline()
            // 编码 是出站操作 将消息编写二进制
            .addLast(new RpcEncoder<>())
            // 解码 是入站操作 将二进制解码成消息
            .addLast(new RpcDecoder())
            // 接收响应 入站操作
            .addLast(handler);
    }
});