kl个人博客 首页>>架构/杂谈>>给gRPC-spring-boot-starter一个pr的说明

给gRPC-spring-boot-starter一个pr的说明

给gRPC-spring-boot-starter一个pr的说明

前言

为了更好的说明给gRPC-spring-boot-starter项目提交bug修复的pr的原因,解答作者的问题。以博文的形式记录了整个过程的上下文,目前pr未合并还在沟通处理中,希望此博文可以更清楚描述问题

pr地址:https://github.com/yidongnan/grpc-spring-boot-starter/pull/454

gRPC-spring-boot-starter是什么?

这是一个spring-boot-starter项目,用来在spring boot框架下,快速便捷的使用grpc技术,开箱即用。它提供如下等功能特性:

  • 在 spring boot 应用中,通过@GrpcService自动配置并运行一个嵌入式的 gRPC 服务。
  • 使用@GrpcClient自动创建和管理您的 gRPC Channels 和 stubs
  • 支持Spring Cloud(向ConsulEurekaNacos注册服务并获取 gRPC 服务端信息)
  • 支持Spring Sleuth作为分布式链路跟踪解决方案(如果brave-instrument-grpc存在)
  • 支持全局和自定义的 gRPC 服务端/客户端拦截器
  • 支持Spring-Security
  • 支持metric (基于micrometer/actuator)
  • 也适用于 (non-shaded) grpc-netty

选型gRPC-spring-boot-starter

博主新入职公司接手的项目采用grpc做微服务通讯框架,项目底层框架采用的spring boot,然后grpc的使用是纯手工配置的,代码写起来比较繁琐, 而且这种繁琐的模板化代码充斥在每个采用了grpc的微服务项目里。所以技术选型后找到了gRPC-spring-boot-starter 这个开源项目,这个项目代码质量不错,非常规范,文档也比较齐全。但是鉴于之前工作经验遇到过开源项目的问题(博主选型的原则,如果有合适的轮子,就摸透这个轮子,然后基于这个轮子二开,没有就自己造一个轮子),而且一般解决周期比较长,所以 最后,我们没有直接采用他们的发行包,而是fork了项目后,打算自己维护。正因为如此,才为后面迅速解决问题上线成为可能。也验证了二开这个选择是正确的。

bug出现,grpc未优雅下线

风风火火重构了所有代码,全部换成gRPC-spring-boot-starter后就上线了,上线后一切都非常好,但是项目在第二次需求上线投产时发生了一些问题。 这个时候还不确定是切换grpc实现导致的问题,现象就是,线上出现了大量的请求异常。上线完成后,异常就消失了。后面每次滚动更新都会出现类似的异常。 这个时候就很容易联系到是否切换grpc实现后,grpc未优雅下线,导致滚动更新时,大量的进行中的请求未正常处理,导致这部分流量异常?因为我们线上 流量比较大,几乎每时每刻都有大量请求,所以我们要求线上服务必须支持无缝滚动更新。如果流量比较小,这个问题可能就不会暴露出来,这也解释了之前和同事讨论的点,为什么这么明显的问题没有被及早的发现。不过都目前为止,这一切都只是猜测,真相继续往下。

定位bug,寻找真实原因

有了上面的猜测,直接找到了gRPC-spring-boot-starter管理维护GrpcServer生命周期的类GrpcServerLifecycle,这个类实现了spring的SmartLifecycle接口,这个接口是用来注册SpringContextShutdownHook的钩子用的,它的实现如下:

@Slf4j
public class GrpcServerLifecycle implements SmartLifecycle {
    private static AtomicInteger serverCounter = new AtomicInteger(-1);

    private volatile Server server;
    private volatile int phase = Integer.MAX_VALUE;
    private final GrpcServerFactory factory;

    public GrpcServerLifecycle(final GrpcServerFactory factory) {
        this.factory = factory;
    }

    @Override
    public void start() {
        try {
            createAndStartGrpcServer();
        } catch (final IOException e) {
            throw new IllegalStateException("Failed to start the grpc server", e);
        }
    }

    @Override
    public void stop() {
        stopAndReleaseGrpcServer();
    }

    @Override
    public void stop(final Runnable callback) {
        stop();
        callback.run();
    }

    @Override
    public boolean isRunning() {
        return this.server != null && !this.server.isShutdown();
    }

    @Override
    public int getPhase() {
        return this.phase;
    }

    @Override
    public boolean isAutoStartup() {
        return true;
    }

    /**
     * Creates and starts the grpc server.
     *
     * @throws IOException If the server is unable to bind the port.
     */
    protected void createAndStartGrpcServer() throws IOException {
        final Server localServer = this.server;
        if (localServer == null) {
            this.server = this.factory.createServer();
            this.server.start();
            log.info("gRPC Server started, listening on address: " + this.factory.getAddress() + ", port: "
                    + this.factory.getPort());

            // Prevent the JVM from shutting down while the server is running
            final Thread awaitThread = new Thread(() -> {
                try {
                    this.server.awaitTermination();
                } catch (final InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }, "grpc-server-container-" + (serverCounter.incrementAndGet()));
            awaitThread.setDaemon(false);
            awaitThread.start();
        }
    }

    /**
     * Initiates an orderly shutdown of the grpc server and releases the references to the server. This call does not
     * wait for the server to be completely shut down.
     */
    protected void stopAndReleaseGrpcServer() {
        final Server localServer = this.server;
        if (localServer != null) {
            localServer.shutdown();
            this.server = null;
            log.info("gRPC server shutdown.");
        }
    }

}

也就是说当spring容器关闭时,会触发ShutdownHook,进而关闭GrpcServer服务,问题就出现在这里,从stopAndReleaseGrpcServer()方法可知,Grpc进行shudown()后,没有进行任何操作,几乎瞬时就返回了,这就导致了进程在收到kill命令时,Grpc的服务会被瞬间回收掉,而不会等待执行中的处理完成,这个判断可以从shutdown()的文档描述中进一步得到确认,如:

  /**
   * Initiates an orderly shutdown in which preexisting calls continue but new calls are rejected.
   * After this call returns, this server has released the listening socket(s) and may be reused by
   * another server.
   *
   * <p>Note that this method will not wait for preexisting calls to finish before returning.
   * {@link #awaitTermination()} or {@link #awaitTermination(long, TimeUnit)} needs to be called to
   * wait for existing calls to finish.
   *
   * @return {@code this} object
   * @since 1.0.0
   */
  public abstract Server shutdown();

文档指出,调用shutdown()后,不在接收新的请求流量,进行中的请求会继续处理完成,但是请注意,它不会等待现有的调用请求完成,必须使用awaitTermination()方法等待请求完成,也就是说,这里处理关闭的逻辑里,缺少了awaitTermination()等待处理中的请求完成的逻辑。

模拟环境,反复验证

验证方法:

这个场景的问题非常容易验证,只需要在server端模拟业务阻塞耗时长一点,然后kill掉java进程,看程序是否会立刻被kill。正常优雅下线关闭的话,会等待阻塞的时间后进程kill。否则就会出现不管业务阻塞多长时间,进程都会立马kill。

验证定位的bug

先验证下是否如上面所说,不加awaitTermination()时,进程是否立马就死了。直接使用gRPC-spring-boot-starter里自带的demo程序,在server端的方法里加上如下模拟业务执行耗时的代码:

@GrpcService public class GrpcServerService extends SimpleGrpc.SimpleImplBase {

    @Override
  public void sayHello(HelloRequest req, StreamObserver<HelloReply> responseObserver) {
        HelloReply reply = HelloReply._newBuilder_().setMessage("Hello ==> " \+ req.getName()).build();
 try {
            System._err_.println("收到请求,阻塞等待");
  TimeUnit._MINUTES_.sleep(1);
  System._err_.println("阻塞完成,请求结束");
  } catch (InterruptedException e) {
            e.printStackTrace();
  }
        responseObserver.onNext(reply);
  responseObserver.onCompleted();
  }
}

上面代码模拟的执行一分钟的方法,然后触发grpc client调用。接着找到server端的进程号,直接kill掉。发现进程确实立马就kill了。继续加大阻塞的时间,从一分钟加大到六分钟,重复测试,还是立马就kill掉了,没有任何的等待。

验证修复后的效果

先将上面的代码修复下,正确的关闭逻辑应该如下,在Grpc发出shutdown指令后,阻塞等待所有请求正常结束,同时,这里阻塞也会夯住主进程不会里面挂掉。

    protected void stopAndReleaseGrpcServer() {
        final Server localServer = this.server;
        if (localServer != null) {
            localServer.shutdown();
            try {
                this.server.awaitTermination();
            } catch (final InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            this.server = null;
            log.info("gRPC server shutdown.");
        }
    }

同样,如上述步骤验证,当kill掉java进程后,此时java进程并没有立马就被kill,而是被awaitTermination()阻塞住了线程,直到业务方法中模拟的业务阻塞结束后,java进程才被kill掉,这正是我们想要达到的优雅下线关闭的效果。被kill时的,线程堆栈如下:

即使被kill了,还是能打印如下的日志【阻塞完成,请求结束】,进一步验证了修复后确实解决了问题:

kl个人博客