自研RPC框架之优雅启停

自研RPC框架之优雅启停

1.优雅关闭

当服务提供方要上线的时候，一般是通过部署系统完成实例重启。在这个过程中，服务提供方的团队并不会事先告诉调用方他们需要操作哪些机器，从而让调用方去事先切走流量。而对调用方来说，它也无法预测到服务提供方要对哪些机器重启上线，因此负载均衡就有可能把要正在重启的机器选出来，这样就会导致把请求发送到正在重启中的机器里面，从而导致调用方不能拿到正确的响应结果。

img

在服务重启的时候，对于调用方来说，这时候可能会存在以下几种情况：

  1. 调用方发请求前，目标服务已经下线。我们的RPC框架中会通过ZooKeeper注册中心的Watcher机制进行服务下线的通知，将下线服务节点从健康列表删除，自然也就不会被负载均衡选中。
  2. 调用方发请求的时候，目标服务正在关闭。此时服务虽然正在关闭，但Watcher通知还没有达到，因此调用方并不知道它正在关闭，而且两者之间的连接也没断开，所以这个节点还会存在健康列表里面，因此该节点就有一定概率会被负载均衡选中。

知道了原因，问题就很好解决了。因为服务提供方已经开始进入关闭流程，那么很多对象就可能已经被销毁了，关闭后再收到的请求按照正常业务请求来处理，肯定是没法保证能处理的。所以我们可以在关闭的时候，设置一个请求“挡板”，挡板的作用就是告诉调用方，我已经开始进入关闭流程了，我不能再处理你这个请求了。

基于这个思路，我们可以这么处理：当服务提供方正在关闭，如果这之后还收到了新的业务请求，服务提供方直接返回一个特定的异常给调用方（比如 ShutdownException）。这个异常就是告诉调用方“我已经收到这个请求了，但是我正在关闭，并没有处理这个请求”，然后调用方收到这个异常响应后，RPC 框架把这个节点从健康列表挪出，并把请求自动重试到其他节点，因为这个请求是没有被服务提供方处理过，所以可以安全地重试到其他节点，这样就可以实现对业务无损。但如果只是靠等待被动调用，就会让这个关闭过程整体有点漫长。因为有的调用方那个时刻没有业务请求，就不能及时地通知调用方了，所以我们可以加上主动通知流程(其实在我们的RPC框架里面可以看作Watcher通知)，这样既可以保证实时性，也可以避免通知失败的情况。

如果进程结束过快会造成这些请求还没有来得及应答，同时调用方会也会抛出异常。为了尽可能地完成正在处理的请求，首先我们要把这些请求识别出来。这就好比日常生活中，我们经常看见停车场指示牌上提示还有多少剩余车位，这个是如何做到的呢？如果仔细观察一下，你就会发现它是每进入一辆车，剩余车位就减一，每出来一辆车，剩余车位就加一。我们也可以利用这个原理在服务对象加上引用计数器，每开始处理请求之前加一，完成请求处理减一，通过该计数器我们就可以快速判断是否有正在处理的请求。

服务对象在关闭过程中，会拒绝新的请求，同时根据引用计数器等待正在处理的请求全部结束之后才会真正关闭。但考虑到有些业务请求可能处理时间长，或者存在被挂住的情况，为了避免一直等待造成应用无法正常退出，我们可以在整个 ShutdownHook 里面，加上超时时间控制，当超过了指定时间没有结束，则强制退出应用。超时时间我建议可以设定成 10s，基本可以确保请求都处理完了。整个流程如下图所示。

img

image-20240210230722014

服务端启动的时候通过Runtime.getRuntime().addShutdownHook我们可以添加自己的关闭钩子，关闭钩子负责优雅关机的实现，当服务关闭时，我们会直接通知注册中心进行服务下线(进而会触发客户端注册的Watcher通知，完成下线服务从健康列表的删除)，然后开启请求挡板，表示关闭期间再来的请求直接拒绝服务，返回特定的响应即可，最后为了尽量保证正常的请求能够处理完毕需要延迟等待所有正常请求处理结束，这是通过请求计数器完成的，当然我们也不可能无限等待，一般来说10s是一个不错的选择。

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
public class OpenrpcShutdownHook extends Thread {
    @Override
    public void run() {
        // 优雅关机
        // 1.服务注销
        OpenrpcBootStrap.configuration.getRegistry().logOff();
        // 2.开启请求挡板
        ShutdownHolder.baffle.set(true);
        // 3.阻塞等待正常请求处理完毕(最多等待10s)
        long start = System.currentTimeMillis();
        while (true) {
            if (ShutdownHolder.normalRequestCount.sum() == 0 || System.currentTimeMillis() - start >= 10000) {
                break;
            }
            try {
                TimeUnit.MILLISECONDS.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

public class ShutdownHolder {
    // 请求挡板，正常情况下为false，服务关闭期间为true
    public static AtomicBoolean baffle = new AtomicBoolean(false);
    // 待处理的正常请求剩余数量
    public static LongAdder normalRequestCount = new LongAdder();
}

那么请求挡板和请求计数器发生的位置是哪里呢？其实正是我们服务端的方法执行处理器：

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
public class MethodCallHandler extends SimpleChannelInboundHandler<OpenrpcRequest> {
    private static final Logger log = LoggerFactory.getLogger(MethodCallHandler.class);
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, OpenrpcRequest msg) throws Exception {
        // 获取请求载荷
        RequestPayload payload = msg.getRequestPayload();
        // 服务关闭期间不处理任何新到的请求
        if (ShutdownHolder.baffle.get()) {
            log.warn("服务正在关闭中:{}", msg.getRequestId());
            OpenrpcResponse response = new OpenrpcResponse(msg.getRequestId(), ResponseCode.CLOSING.getCode(), msg.getCompressType(), msg.getSerializeType(), "");
            ctx.channel().writeAndFlush(response);
            return;
        }
        // 请求限流处理
        Limiter limiter = OpenrpcBootStrap.configuration.getLimiter();
        if (limiter.limit()) {
            log.warn("请求负载过高,限流处理:{}", msg.getRequestId());
            OpenrpcResponse response = new OpenrpcResponse(msg.getRequestId(), ResponseCode.LIMITING.getCode(), msg.getCompressType(), msg.getSerializeType(), "");
            ctx.channel().writeAndFlush(response);
            return;
        }
        // 正常请求计数
        ShutdownHolder.normalRequestCount.increment();
        // 执行匹配方法
        Object result = callTargetMethod(payload);
        if (result == null) {
            OpenrpcResponse response = new OpenrpcResponse(msg.getRequestId(), ResponseCode.FAIL.getCode(), msg.getCompressType(), msg.getSerializeType(), "");
            ctx.channel().writeAndFlush(response);
            log.warn("请求处理失败:{}", msg.getRequestId());
        }
        log.info("服务端的请求方法调用工作已完成:{}", msg.getRequestId());
        // 封装响应
        OpenrpcResponse response = new OpenrpcResponse(msg.getRequestId(), ResponseCode.SUCCESS.getCode(), msg.getCompressType(), msg.getSerializeType(), result);
        // 正常请求计数
        ShutdownHolder.normalRequestCount.decrement();
        // 返回响应，开始进入服务端出站的流水线处理器
        ctx.channel().writeAndFlush(response);
    }
	......
}

2.优雅启动

运行了一段时间后的应用，执行速度会比刚启动的应用更快。这是因为在Java里面，在运行过程中，JVM虚拟机会把高频的代码编译成机器码，被加载过的类也会被缓存到JVM缓存中，再次使用的时候不会触发临时加载，这样就使得“热点”代码的执行不用每次都通过解释，从而提升执行速度。但是这些“临时数据”，都在我们应用重启后就消失了。重启后的这些“红利”没有了之后，如果让我们刚启动的应用就承担像停机前一样的流量，这会使应用在启动之初就处于高负载状态，从而导致调用方过来的请求可能出现大面积超时，进而对线上业务产生损害行为。

这涉及到RPC里面的一个实用功能——启动预热，简单来说，就是让刚启动的服务提供方应用不承担全部的流量，而是让它被调用的次数随着时间的移动慢慢增加，最终让流量缓和地增加到跟已经运行一段时间后的水平一样。

我们应用启动的时候都是通过main入口，然后顺序加载各种相关依赖的类。以Spring应用启动为例，在加载的过程中，Spring容器会顺序加载Spring Bean，如果某个Bean是RPC服务的话，我们不光要把它注册到Spring-BeanFactory里面去，还要把这个Bean对应的接口注册到注册中心。注册中心在收到新上线的服务提供方地址的时候，会把这个地址推送到调用方应用内存中；当调用方收到这个服务提供方地址的时候，就会去建立连接发请求。

但这时候是不是存在服务提供方可能并没有启动完成的情况？因为服务提供方应用可能还在加载其它的Bean。对于调用方来说，只要获取到了服务提供方的IP，就有可能发起RPC调用，但如果这时候服务提供方没有启动完成的话，就会导致调用失败，从而使业务受损。

在解决问题前，我们先看下出现上述问题的根本原因。==这是因为服务提供方应用在没有启动完成的时候，调用方的请求就过来了，而调用方请求过来的原因是，服务提供方应用在启动过程中把解析到的RPC服务注册到了注册中心，这就导致在后续加载没有完成的情况下服务提供方的地址就被服务调用方感知到了。==

这样的话，其实我们就可以把接口注册到注册中心的时间挪到应用启动完成后。具体的做法就是在应用启动加载、解析Bean的时候，如果遇到了RPC服务的Bean，只先把这个Bean注册到Spring-BeanFactory里面去，而并不把这个Bean对应的接口注册到注册中心，只有等应用启动完成后，才把接口注册到注册中心用于服务发现，从而实现让服务调用方延迟获取到服务提供方地址。

这样是可以保证应用在启动完后才开始接入流量的，但其实这样做，我们还是没有实现最开始的目标。因为这时候应用虽然启动完成了，但并没有执行相关的业务代码，所以JVM内存里面还是冷的。如果这时候大量请求过来，还是会导致整个应用在高负载模式下运行，从而导致不能及时地返回请求结果。而且在实际业务中，一个服务的内部业务逻辑一般会依赖其它资源的，比如缓存数据。如果我们能在服务正式提供服务前，先完成缓存的初始化操作，而不是等请求来了之后才去加载，我们就可以降低重启后第一次请求出错的概率。

我们还是需要利用服务提供方把接口注册到注册中心的那段时间。我们可以在服务提供方应用启动后，接口注册到注册中心前，预留一个Hook 过程，让用户可以实现可扩展的Hook逻辑。用户可以在Hook里面模拟调用逻辑，从而使JVM指令能够预热起来，并且用户也可以在Hook里面事先预加载一些资源，只有等所有的资源都加载完成后，最后才把接口注册到注册中心。

整个应用启动过程如下图所示：

img

image-20240210232637263

1
2
3
4
5
public interface OpenrpcStartupHook {
    default void preheat() {
        System.out.println("启动预热过程...");
    }
}