Dubbo扩展支持自适应等待无损下线

vlambda
2022-04-20

Dubbo扩展支持自适应等待无损下线

无损上下线是服务治理不可忽视的问题，在应⽤上下线发布过程中，如果上下线不平滑，就会出现短时间的服务调⽤报错，如连接被拒绝（ Connection refused）、请求超时或请求异常。

请求超时、请求异常发生在服务提供者上线时，由于过早暴露服务，请求进来时，可能应用还未初始化完成，如中间件的初始化。连接被拒绝、等待响应超时、请求异常发生在服务下线，如果中间件的销毁早于Dubbo就会出现请求异常；如果请求未写入IO通道就关闭连接，就会导致服务消费者等待响应超时；如果服务消费者感知提供者下线有延迟，就会导致延迟的这段时间内，被路由到已下线提供者节点的请求都抛连接被拒绝异常。

对于无损上线（平滑上线），Dubbo提供了延迟注册的解决方案，可以结合延迟初始化使用。在Spring容器初始化阶段，我们先将服务提供者扫描出来，不影响Spring对提供者实现Bean的生命周期处理，等待Spring容器初始化完成之后，通过监听Spring容器初始化完成事件，再将扫描出来的服务提供者注册到注册中心，此时还可以结合Dubbo的延迟注册功能使用，避免一些中间件组件也是在这个时机才初始化。

对于无损下线（平滑下线），Dubbo也提供了ShutdownHook的支持，但这个实现比较简陋。如果使用Dubbo的ShutdownHook，会导致正在处理中的请求（处理ing）无法正常完成处理和响应。

为解决此问题，我们可实现自适应等待无损下线，移除Dubbo注册的ShutdownHook，自己注册一个ShutdownHook，并在这个ShutdownHook中，先是将此服务提供者节点从注册中心摘除，此时还是能够继续接收请求的，然后休眠等待所有正在处理中的请求都完成，并且响应给消费者后，再销毁协议（如http协议的jetty容器），不再接收和处理请求。

这个过程是：先从注册中心注销->继续接收和处理请求->处理完最后的请求后销毁协议->不再接收请求和处理请求。当服务提供者处于“继续接收和处理请求”这个阶段时，消费者会陆续感知到此节点已经下线，后续不再发请求。

移除Dubbo的ShutdownHook需要确保在Dubbo调用addShutdownHook之后。移除的代码如下。

 
   
   
 
  
    
    
  // 移除dubbo的钩子，实现无损下线需要，避免接收到kill信号量就把协议销毁了
  
    
    
  

  
    
    
  // 注册的地方@see com.alibaba.dubbo.config.AbstractConfig#static{}
  
    
    
  
Runtime.getRuntime().removeShutdownHook(DubboShutdownHook.getDubboShutdownHook());

然后注册自己的ShutdownHook，例如。

 
   
   
 
  
    
    
  static {
  
    
    
  
    Runtime.getRuntime().addShutdownHook(
  
    
    
  new Thread(
  
    
    
  new Runnable() {
  
    
    
  
        
  
    
    
  @Override
  
    
    
  
        
  
    
    
  public void run() {
  
    
    
  
            shutdownIfNeed();
  
    
    
  
        }
  
    
    
  
    }));
  
    
    
  
}
  
    
    
  

  
    
    
  

  
    
    
  private synchronized static void shutdownIfNeed() {
  
    
    
  
    
  
    
    
  //......
  
    
    
  
    
  
    
    
  // 先从注册中心摘除
  
    
    
  
    
  
    
    
  for (ServiceConfig<Object> serviceConfig : SERVICE_MAP.values()) {
  
    
    
  
        serviceConfig.unexport();
  
    
    
  
    }
  
    
    
  
    
  
    
    
  // 无损下线等待
  
    
    
  
    LosslessOfflineSupper.losslessOffline();
  
    
    
  
    
  
    
    
  // double unexport，销毁协议
  
    
    
  
    DubboShutdownHook.getDubboShutdownHook().destroyAll();
  
    
    
  
}

实现自适应等待，可通过Filter扩展点，添加一个负责统计正在处理的请求数的Filter，例如。

 
   
   
 
  
    
    
  @Activate(group = Constants.PROVIDER)
  
    
    
  

  
    
    
  public 
  
    
    
  class LosslessOfflineProviderFilter implements Filter {
  
    
    
  
    
  
    
    
  @Override
  
    
    
  
    
  
    
    
  public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
  
    
    
  
        LosslessOfflineSupper.incRequest();
  
    
    
  
        
  
    
    
  try {
  
    
    
  
            
  
    
    
  return invoker.invoke(invocation);
  
    
    
  
        } 
  
    
    
  finally {
  
    
    
  
            LosslessOfflineSupper.decRequest();
  
    
    
  
        }
  
    
    
  
    }
  
    
    
  
}

统计逻辑的实现很简单，使用AtomicLong统计即可，在 invoker.invoke调用之前自增处理中的请求数，在 invoker.invoke调用之后自减处理中的请求数。

等待逻辑的实现如下。

 
   
   
 
  
    
    
  public static void losslessOffline() {
  
    
    
  
    
  
    
    
  while (REQ_CNT.get() > 
  
    
    
  0) {
  
    
    
  
        
  
    
    
  try {
  
    
    
  
            Thread.sleep(
  
    
    
  100);
  
    
    
  
        } 
  
    
    
  catch (InterruptedException e) {
  
    
    
  
            
  
    
    
  //
  
    
    
  
        }
  
    
    
  
    }
  
    
    
  
    
  
    
    
  // 响应给服务消费者可能还需要点时间
  
    
    
  
    
  
    
    
  try {
  
    
    
  
        Thread.sleep(
  
    
    
  5000);
  
    
    
  
    } 
  
    
    
  catch (InterruptedException e) {
  
    
    
  
        
  
    
    
  //
  
    
    
  
    }
  
    
    
  
}

最后再休眠几秒，是为了避免响应没写给消费者就关闭了连接。

需要注意，此方案依然解决不了中间件组件通过ShutdownHook在Dubbo销毁之前先销毁的问题。由于ShutdownHook的 无序和 异步特性，如果中间件组件也注册了ShutdownHook，且这些ShutdownHook在Dubbo的ShutdownHook之前已经执行完了，如果还有请求进来，这些请求就无法被正常处理，可能还会导致产生脏数据。如发送kafka成功后，写mysql失败，因为mysql连接池这时候已经销毁，那么请求处理失败了，但已经发送kafka成功的消息却无法撤回。

当然，如果是将Dubbo整合到Spring项目中，建议是使用Spring的事件监听完成shutdown操作，避免Spring在dubbo之前shutdown，导致一些bean的销毁方法被调用，无法再处理业务逻辑。

只需要添加一个Bean，实现ApplicationListener接口，并指定泛型参数类型为ContextClosedEvent即可监听Spring的Shutdown事件。并且ApplicationListener是支持使用Spring的注解排序的，这样能指定将Dubbo的ApplicationListener排在最前面，例如。

 
   
   
 
  
    
    
  @Configuration
  
    
    
  

  
    
    
  public 
  
    
    
  class DubboApplicationListener implements ApplicationListener<ContextClosedEvent>, Ordered{
  
    
    
  
    
  
    
    
  @Override
  
    
    
  
    
  
    
    
  public int getOrder() {
  
    
    
  
        
  
    
    
  return Ordered.LOWEST_PRECEDENCE;
  
    
    
  
    }
  
    
    
  
    
  
    
    
  @Override
  
    
    
  
    
  
    
    
  public void onApplicationEvent(ContextClosedEvent event) {
  
    
    
  
        
  
    
    
  // do 
  
    
    
  
    }
  
    
    
  
}

Spring的ContextClosedEvent事件实际也是通过注册JVM ShutdownHook，然后调用ApplicationContext的doClose方法，在doClose方法中发出的事件。

Spring保证了在调用ApplicationListener的onApplicationEvent方法之后，才会执行销毁bean的逻辑。但Spring不会默认add这个ShutdownHook，需要我们手动调用registerShutdownHook方法才会生效。

 
   
   
 
  
    
    
  org.springframework.context.support.AbstractApplicationContext#registerShutdownHook

如果是SpringBoot应用，无需手动调用，SpringBoot已经做了封装，在SpringApplication的refreshContext方法中调用了ApplicationContext的registerShutdownHook方法。

无论是使用Runtime.getRuntime().addShutdownHook，还是Spring的ContextClosedEvent事件，要实现应用平滑下线，单Dubbo是不够的，需要整个技术栈提供支持，比如，约定都通过Spring的ContextClosedEvent实现shutdown操作，并且约定shutdown顺序。

还有另外一种取巧的方法，通过反射注册一个更高优先级的Hook，可以让该Hook在所有调用Runtime.getRuntime().addShutdownHook方法注册的ShutdownHook之前执行，经验证，该方案可行。

 
   
   
 
  
    
    
  /**
 * @see java.io.Console使用了slot=0，要确保没有地方用到java.io.Console
 */
  
    
    
  

  
    
    
  static {
  
    
    
  
    
  
    
    
  try {
  
    
    
  
        Class<?> sc = Class.forName(
  
    
    
  "java.lang.Shutdown");
  
    
    
  
        Method method = sc.getDeclaredMethod(
  
    
    
  "add", 
  
    
    
  int.class, 
  
    
    
  boolean.class, Runnable.class);
  
    
    
  
        method.setAccessible(
  
    
    
  true);
  
    
    
  
        
  
    
    
  // 插入在ApplicationShutdownHooks之前，前提是slot=0没被占用
  
    
    
  
        method.invoke(
  
    
    
  null, 
  
    
    
  0, 
  
    
    
  false, 
  
    
    
  new Runnable() {
  
    
    
  
            
  
    
    
  @Override
  
    
    
  
            
  
    
    
  public void run() {
  
    
    
  
                
  
    
    
  // do 
  
    
    
  
            }
  
    
    
  
        });
  
    
    
  
    } 
  
    
    
  catch (Throwable e) {
  
    
    
  
    }
  
    
    
  
}

需要注意的是，Shutdown的add方法，传递的slot只能是0~9,并且1已经被JDK实现Runtime.getRuntime().addShutdownHook使用了，0和2也被使用了，其中，只要没有地方触发java.io.Console类初始化，0就可以使用，否则会导致进程启动不起来。

这种方法注册的Hook是会阻塞后面的Hook的执行的，而Runtime.getRuntime().addShutdownHook注册的ShutdownHook不仅无法控制排序，每个ShutdownHook都是一个线程，也无法控制ShutdownHook-A执行完之后再到ShutdownHook-B的执行顺序。