gRPC Server 端请求处理流程

b.childHandler(new ChannelInitializer<Channel>() {
    @Override
    public void initChannel(Channel ch) {
        // 构建基于 Netty 的 ServerTransport
        NettyServerTransport transport = new NettyServerTransport(/*...*/);
        ServerTransportListener transportListener;

        synchronized (NettyServer.this) {
            // 调用监听器回调，Transport 创建事件
            transportListener = listener.transportCreated(transport);
        }

        // 启动监听器
        transport.start(transportListener);
        ChannelFutureListener loopReleaser = new LoopReleaser();
        channelDone.addListener(loopReleaser);
        ch.closeFuture().addListener(loopReleaser);
    }
});

public void start(ServerTransportListener listener) {
    this.listener = listener;

    // 为 pipeline 创建 Netty Handler
    grpcHandler = createHandler(listener, channelUnused);

    // 创建 Handler
    ChannelHandler negotiationHandler = protocolNegotiator.newHandler(grpcHandler);
    ChannelHandler bufferingHandler = new WriteBufferingAndExceptionHandler(negotiationHandler);

    // 添加监听器
    ChannelFutureListener terminationNotifier = new TerminationNotifier();
    channelUnused.addListener(terminationNotifier);
    channel.closeFuture().addListener(terminationNotifier);

    channel.pipeline().addLast(bufferingHandler);
}

public void onSettingsRead(ChannelHandlerContext ctx, Http2Settings settings) {
    if (firstSettings) {
        firstSettings = false;
        // 通知 Transport ready
        attributes = transportListener.transportReady(negotiationAttributes);
    }
}

public Attributes transportReady(Attributes attributes) {
    // 如果有握手超时回调，则取消
    handshakeTimeoutFuture.cancel(false);
    handshakeTimeoutFuture = null;

    // 遍历 TransportFilter，通知 ready 事件并获取 attributes
    for (ServerTransportFilter filter : transportFilters) {
        attributes = Preconditions.checkNotNull(filter.transportReady(attributes), "Filter %s returned null", filter);
    }
    this.attributes = attributes;
    return attributes;
}

public void onHeadersRead(ChannelHandlerContext ctx, int streamId, Http2Headers headers, int streamDependency, short weight, boolean exclusive, int padding, boolean endStream) throws Http2Exception {
    if (NettyServerHandler.this.keepAliveManager != null) {
        NettyServerHandler.this.keepAliveManager.onDataReceived();
    }

    NettyServerHandler.this.onHeadersRead(ctx, streamId, headers);
}

public void onHeadersRead(ChannelHandlerContext ctx,
                          int streamId,
                          Http2Headers headers,
                          int streamDependency,
                          short weight,
                          boolean exclusive,
                          int padding,
                          boolean endStream) throws Http2Exception {
    if (keepAliveManager != null) {
        keepAliveManager.onDataReceived();
    }
    // 最终会创建流并出发流创建事件
    NettyServerHandler.this.onHeadersRead(ctx, streamId, headers);
}

private void onHeadersRead(ChannelHandlerContext ctx, int streamId, Http2Headers headers) throws Http2Exception {
  try {
    // 删除斜杠获取方法限定名称
    CharSequence path = headers.path();

    // 方法限定名，即包含服务名和方法名
    String method = path.subSequence(1, path.length()).toString();
    // 获取 HTTP 流
    Http2Stream http2Stream = requireHttp2Stream(streamId);

    // 将 header 转为 metadata
    Metadata metadata = Utils.convertHeaders(headers);
    // 创建支持统计的上下文
    StatsTraceContext statsTraceCtx = StatsTraceContext.newServerContext(streamTracerFactories, method, metadata);

    // 创建流的声明
    NettyServerStream.TransportState state = new NettyServerStream.TransportState(
            this,
            ctx.channel().eventLoop(),
            http2Stream,
            maxMessageSize,
            statsTraceCtx,
            transportTracer,
            method);

    try {
      // 获取请求的 authority
      String authority = getOrUpdateAuthority((AsciiString) headers.authority());
      // 创建 Server 端的流
      NettyServerStream stream = new NettyServerStream(ctx.channel(),
              state,
              attributes,
              authority,
              statsTraceCtx,
              transportTracer);

      // 触发监听器，通知流创建事件，查找相应处理器，开始处理流，会提交 StreamCreated 任务到线程池中
      transportListener.streamCreated(stream, method, metadata);
      // 会提交 OnReady 任务到线程池中，通知 Stream Ready
      state.onStreamAllocated();
      http2Stream.setProperty(streamKey, state);
    }
  } catch (Exception e) {
    logger.log(Level.WARNING, "Exception in onHeadersRead()", e);
    throw newStreamException(streamId, e);
  }
}

transportListener.streamCreated(stream, method, metadata);

private void streamCreatedInternal(final ServerStream stream,
                                   final String methodName,
                                   final Metadata headers,
                                   final Tag tag) {
    final Executor wrappedExecutor;

    if (executor == directExecutor()) {
        wrappedExecutor = new SerializeReentrantCallsDirectExecutor();
        stream.optimizeForDirectExecutor();
    } else {
        // 否则使用指定的 Executor 执行
        wrappedExecutor = new SerializingExecutor(executor);
    }

    // 创建可以取消的上下文
    final Context.CancellableContext context = createContext(headers, statsTraceCtx);

    // 流事件监听器，处理流的所有生命周期事件
    final JumpToApplicationThreadServerStreamListener jumpListener = new JumpToApplicationThreadServerStreamListener(wrappedExecutor, executor, stream, context, tag);
    stream.setListener(jumpListener);

    // 提交流创建任务
    wrappedExecutor.execute(new StreamCreated());
}

state.onStreamAllocated();

protected void onStreamAllocated() {
    checkState(listener() != null);
    synchronized (onReadyLock) {
        checkState(!allocated, "Already allocated");
        allocated = true;
    }
    notifyIfReady();
}

public void onReady() {
    try {
        callExecutor.execute(new OnReady());
    }
}

// 流创建任务处理
    final class StreamCreated extends ContextRunnable {
 
        private void runInternal() {
            ServerStreamListener listener = NOOP_LISTENER;
            try {
                // 根据方法名称获取方法定义
                ServerMethodDefinition<?, ?> method = registry.lookupMethod(methodName);
                // 如果没有则从回退的方法注册器中查找
                if (method == null) {
                    method = fallbackRegistry.lookupMethod(methodName, stream.getAuthority());
                }
                // 如果没有则方法不存在，返回 UNIMPLEMENTED，关闭流，取消上下文
                if (method == null) {
                    Status status = Status.UNIMPLEMENTED.withDescription("Method not found: " + methodName);
                    stream.close(status, new Metadata());
                    context.cancel(null);
                    return;
                }
                // 如果方法存在，则开始调用
                listener = startCall(stream, methodName, method, headers, context, statsTraceCtx, tag);
            } catch (Throwable t) {
                stream.close(Status.fromThrowable(t), new Metadata());
                context.cancel(null);
                throw t;
            } finally {
                 jumpListener.setListener(listener);
            }
           
            final class ServerStreamCancellationListener implements Context.CancellationListener {
                @Override
                public void cancelled(Context context) {
                    Status status = statusFromCancelled(context);
                    if (DEADLINE_EXCEEDED.getCode().equals(status.getCode())) {
                        stream.cancel(status);
                    }
                 }
            }
            context.addListener(new ServerStreamCancellationListener(), directExecutor());
         }
    }

private <ReqT, RespT> ServerStreamListener startCall(ServerStream stream,
                                                     String fullMethodName,
                                                     ServerMethodDefinition<ReqT, RespT> methodDef,
                                                     Metadata headers,
                                                     Context.CancellableContext context,
                                                     StatsTraceContext statsTraceCtx,
                                                     Tag tag) {
    // 从方法描述获取调用处理器
    ServerCallHandler<ReqT, RespT> handler = methodDef.getServerCallHandler();
    // 遍历拦截器，为处理器添加拦截器
    for (ServerInterceptor interceptor : interceptors) {
        handler = InternalServerInterceptors.interceptCallHandler(interceptor, handler);
    }
    // 使用添加了拦截器后的处理器创建新的方法定义
    ServerMethodDefinition<ReqT, RespT> interceptedDef = methodDef.withServerCallHandler(handler);

    // 处理封装后的调用
    return startWrappedCall(fullMethodName, wMethodDef, stream, headers, context, tag);
}

private <WReqT, WRespT> ServerStreamListener startWrappedCall(String fullMethodName,
                                                              ServerMethodDefinition<WReqT, WRespT> methodDef,
                                                              ServerStream stream,
                                                              Metadata headers,
                                                              Context.CancellableContext context,
                                                              Tag tag) {

    // 创建请求处理器
    ServerCallImpl<WReqT, WRespT> call = new ServerCallImpl<>(stream,
            methodDef.getMethodDescriptor(),
            headers,
            context,
            decompressorRegistry,
            compressorRegistry,
            serverCallTracer,
            tag);

    // 调用方法处理器，真正调用实现逻辑的方法
    ServerCall.Listener<WReqT> listener = methodDef.getServerCallHandler().startCall(call, headers);
    if (listener == null) {
        throw new NullPointerException("startCall() returned a null listener for method " + fullMethodName);
    }

    // 根据调用监听器创建新的流监听器
    return call.newServerStreamListener(listener);
}

public ServerCall.Listener<ReqT> startCall(ServerCall<ReqT, RespT> call, Metadata headers) {
    // 创建响应处理器
    ServerCallStreamObserverImpl<ReqT, RespT> responseObserver = new ServerCallStreamObserverImpl<>(call);
    // 会调用 io.grpc.internal.AbstractStream#request 方法获取消息
    call.request(2);
    // 返回监听器
    return new UnaryServerCallListener(responseObserver, call);
}

public final void request(int numMessages) {
    transportState().requestMessagesFromDeframer(numMessages);
}

private void requestMessagesFromDeframer(final int numMessages) {
    // 如果不是线程安全的解帧器，则由 Transport 的线程执行
    class RequestRunnable implements Runnable {
        @Override
        public void run() {
            try {
                deframer.request(numMessages);
            } catch (Throwable t) {
                deframeFailed(t);
            }
        }
    }
    runOnTransportThread(new RequestRunnable());
}

final class OnReady extends ContextRunnable {
    OnReady() {
        super(context);
    }

    @Override
    public void runInContext() {
        try {
            // 调用监听器的 ready 事件
            getListener().onReady();
        } catch (Throwable t) {
            internalClose(t);
            throw t;
        }
    }
}

public void onReady() {
    // 将 ready 状态变为 true
    wasReady = true;
    // 如果响应有 readyHandler，则执行
    if (responseObserver.onReadyHandler != null) {
        responseObserver.onReadyHandler.run();
    }
}

class RequestRunnable implements Runnable {
    @Override
    public void run() {
        try {
            deframer.request(numMessages);
        } catch (Throwable t) {
            deframeFailed(t);
        }
    }
}

public void request(int numMessages) {
  if (isClosed()) {
    return;
  }
  pendingDeliveries += numMessages;
  deliver();
}

private void deliver() {
  // 检查投递的状态
  if (inDelivery) {
    return;
  }
  inDelivery = true;
  try {
    // 如果没有停止投递，且有等待投递的消息，且读取成功，则根据相应状态进行处理
    while (!stopDelivery && pendingDeliveries > 0 && readRequiredBytes()) {
      switch (state) {
        case HEADER:
          // 处理 header
          processHeader();
          break;
        case BODY:
          // 处理body
          processBody();
          pendingDeliveries--;
          break;
        default:
          throw new AssertionError("Invalid state: " + state);
      }
    }

    // 如果已经停止投递，则关闭
    if (stopDelivery) {
      close();
      return;
    }

    if (closeWhenComplete && isStalled()) {
      close();
    }
  } finally {
    inDelivery = false;
  }
}

private void processBody() {
  // 读取请求体的流
  InputStream stream = compressedFlag ? getCompressedBody() : getUncompressedBody();
  nextFrame = null;
  // 通知监听器有新的消息
  listener.messagesAvailable(new SingleMessageProducer(stream));

  // 将状态改为处理 header
  state = State.HEADER;
  requiredLength = HEADER_LENGTH;
}

public void messagesAvailable(final MessageProducer producer) {
    try {
        // 执行任务
        callExecutor.execute(new MessagesAvailable());
    }
}

final class MessagesAvailable extends ContextRunnable {

    MessagesAvailable() {
        super(context);
    }

    @Override
    public void runInContext() {
        try {
            // 获取监听器，通知有新的消息
            getListener().messagesAvailable(producer);
        } catch (Throwable t) {
            internalClose(t);
            throw t;
        }
    }
}

private void messagesAvailableInternal(final MessageProducer producer) {
    // 如果调用已经取消了，则关闭生产者
    if (call.cancelled) {
        GrpcUtil.closeQuietly(producer);
        return;
    }

    InputStream message;
    try {
        // 从生产者中获取消息，
        while ((message = producer.next()) != null) {
            try {
                // 将流解析为请求对象，发送给监听器
                listener.onMessage(call.method.parseRequest(message));
            } catch (Throwable t) {
                GrpcUtil.closeQuietly(message);
                throw t;
            }
            message.close();
        }
    } catch (Throwable t) {
        GrpcUtil.closeQuietly(producer);
        Throwables.throwIfUnchecked(t);
        throw new RuntimeException(t);
    }
}

public void onMessage(ReqT request) {
    // 如果已经接收到了一个请求，则返回错误
    if (this.request != null) {
        call.close(Status.INTERNAL.withDescription(TOO_MANY_REQUESTS), new Metadata());
        canInvoke = false;
        return;
    }

    // 延迟执行调用 method.invoke() 直到 onHalfClose() 以确保客户端执行了半关闭
    this.request = request;
}

public void close() {
    if (isClosed()) {
      return;
    }
    boolean hasPartialMessage = nextFrame != null && nextFrame.readableBytes() > 0;
    try {
      if (fullStreamDecompressor != null) {
        hasPartialMessage = hasPartialMessage || fullStreamDecompressor.hasPartialData();
        fullStreamDecompressor.close();
      }
      if (unprocessed != null) {
        unprocessed.close();
      }
      if (nextFrame != null) {
        nextFrame.close();
      }
    } finally {
      fullStreamDecompressor = null;
      unprocessed = null;
      nextFrame = null;
    }
    listener.deframerClosed(hasPartialMessage);
  }

public void deframerClosed(boolean hasPartialMessage) {
    deframerClosed = true;
    // 是否到达流结尾
    if (endOfStream) {
        // 如果不需要立即关闭，且有未完成的消息，返回错误并抛出异常
        if (!immediateCloseRequested && hasPartialMessage) {
            deframeFailed(Status.INTERNAL.withDescription("Encountered end-of-stream mid-frame")
                                         .asRuntimeException());
            deframerClosedTask = null;
            return;
        }
        // 通知半关闭
        listener.halfClosed();
    }
    // 如果有解帧器关闭的任务，则执行
    if (deframerClosedTask != null) {
        deframerClosedTask.run();
        deframerClosedTask = null;
    }
}

public void halfClosed() {
    try {
        callExecutor.execute(new HalfClosed());
    } 
}

final class HalfClosed extends ContextRunnable {
    HalfClosed() {
        super(context);
    }

    @Override
    public void runInContext() {
        try {
            // 调用监听器的半关闭事件
            getListener().halfClosed();
        } catch (Throwable t) {
            internalClose(t);
            throw t;
        }
    }
}

public void onHalfClose() {
    // 如果不能调用则直接返回
    if (!canInvoke) {
        return;
    }

    // 如果请求是 null，则返回错我
    if (request == null) {
        call.close(Status.INTERNAL.withDescription(MISSING_REQUEST), new Metadata());
        return;
    }

    // 执行方法调用
    method.invoke(request, responseObserver);
    // 处理了请求之后将请求置为 null
    request = null;
    // 冻结响应
    responseObserver.freeze();
    // 判断是否 ready
    if (wasReady) {
        // 因为在 halfClose 中调用，错过了来自 Transport 的 onReady 事件，从这里恢复
        // 即在 ready 之后用于执行 onReadyHandler
        onReady();
    }
}

public void invoke(Req request, io.grpc.stub.StreamObserver<Resp> responseObserver) {
    switch (methodId) {
        case METHODID_HOW_ARE_YOU:
            serviceImpl.howAreYou((io.github.helloworlde.HelloMessage) request,
                    (io.grpc.stub.StreamObserver<io.github.helloworlde.HelloResponse>) responseObserver);
            break;
        default:
            throw new AssertionError();
    }
}

public void howAreYou(HelloMessage request, StreamObserver<HelloResponse> responseObserver) {
    responseObserver.onNext(HelloResponse.newBuilder().setResult("Hello : " + request.getMessage()).build());
    responseObserver.onCompleted();
}

public void onNext(RespT response) {
    // 如果已经被取消调用了，则判断是否有取消回调，如果没有则返回取消状态
    if (cancelled) {
        if (onCancelHandler == null) {
            throw Status.CANCELLED.withDescription("call already cancelled").asRuntimeException();
        }
        return;
    }
    // 检查是否已经丢弃或者完成
    checkState(!aborted, "Stream was terminated by error, no further calls are allowed");
    checkState(!completed, "Stream is already completed, no further calls are allowed");
    // 如果还没有发送 header，则发送 header
    if (!sentHeaders) {
        call.sendHeaders(new Metadata());
        // 将发送 header 设置为 true
        sentHeaders = true;
    }
    // 然后发送响应
    call.sendMessage(response);
}

private void sendHeadersInternal(Metadata headers) {
        // 丢弃编码的 key
        headers.discardAll(MESSAGE_ENCODING_KEY);
        // 设置压缩器类型
        headers.put(MESSAGE_ENCODING_KEY, compressor.getMessageEncoding());
        // 为流设置压缩器
        stream.setCompressor(compressor);

        // 丢弃消息编码的 key
        headers.discardAll(MESSAGE_ACCEPT_ENCODING_KEY);
        if (advertisedEncodings.length != 0) {
            headers.put(MESSAGE_ACCEPT_ENCODING_KEY, advertisedEncodings);
        }

        // 将调用 header 状态改为 true
        sendHeadersCalled = true;
        stream.writeHeaders(headers);
    }

public final void writeHeaders(Metadata headers) {
    Preconditions.checkNotNull(headers, "headers");

    headersSent = true;
    abstractServerStreamSink().writeHeaders(headers);
}

private void sendMessageInternal(RespT message) {
    // 检查是否已经发送了 header，和调用是否已经被关闭
    checkState(sendHeadersCalled, "sendHeaders has not been called");
    checkState(!closeCalled, "call is closed");

    // 如果是 UNARY 或者 CLIENT_STREAMING 类型的消息，且已经发送过消息了，则不允许再发送，返回错误状态
    if (method.getType().serverSendsOneMessage() && messageSent) {
        internalClose(Status.INTERNAL.withDescription(TOO_MANY_RESPONSES));
        return;
    }

    // 将发送消息状态改为 true
    messageSent = true;
    try {
        // 将消息序列化为流，写入消息，清空流
        InputStream resp = method.streamResponse(message);
        stream.writeMessage(resp);
        stream.flush();
    } catch (RuntimeException e) {
        close(Status.fromThrowable(e), new Metadata());
    } catch (Error e) {
        close(Status.CANCELLED.withDescription("Server sendMessage() failed with Error"), new Metadata());
        throw e;
    }
}

public final void writeMessage(InputStream message) {
    checkNotNull(message, "message");
    try {
        if (!framer().isClosed()) {
            // 写入消息体
            framer().writePayload(message);
        }
    } finally {
        GrpcUtil.closeQuietly(message);
    }
}

public final void flush() {
    // 如果帧还没有关闭，则清空帧
    if (!framer().isClosed()) {
        framer().flush();
    }
}

public void onCompleted() {
    // 如果已经被取消，则返回取消的状态
    if (cancelled) {
        if (onCancelHandler == null) {
            throw Status.CANCELLED.withDescription("call already cancelled").asRuntimeException();
        }
    } else {
        // 通知请求完成
        call.close(Status.OK, new Metadata());
        // 将完成状态改为 true
        completed = true;
    }
}

private void closeInternal(Status status, Metadata trailers) {
    // 检查是否已经关闭
    checkState(!closeCalled, "call already closed");
    try {
        // 将关闭状态改为 true
        closeCalled = true;

        // 检查状态如果是 OK，且方法类型是 Server 端只能发送一次，且没有发送消息，则返回错误
        if (status.isOk() && method.getType().serverSendsOneMessage() && !messageSent) {
            internalClose(Status.INTERNAL.withDescription(MISSING_RESPONSE));
            return;
        }

        // 关闭流
        stream.close(status, trailers);
    } finally {
        // 统计结果
        serverCallTracer.reportCallEnded(status.isOk());
    }
}

public final void close(Status status, Metadata trailers) {
    // 如果出站的流还未关闭，则将状态改为关闭
    if (!outboundClosed) {
        outboundClosed = true;
        // 从服务端关闭 framer
        endOfMessages();
        // 将响应状态加入到 header 中
        addStatusToTrailers(trailers, status);
        // 安全设置，无需同步，因为访问被严格控制，只有这里设置关闭状态，保证在这里之后读取
        // 给 Transport 设置响应状态
        transportState().setClosedStatus(status);
        // 将响应的 header 信息写入帧中
        abstractServerStreamSink().writeTrailers(trailers, headersSent, status);
    }
}

protected final void endOfMessages() {
    framer().close();
}

abstractServerStreamSink().writeTrailers(trailers, headersSent, status);

public void writeTrailers(Metadata trailers, boolean headersSent, Status status) {
    try {
        Http2Headers http2Trailers = Utils.convertTrailers(trailers, headersSent);
        // 将发送 header 的指令写入到队列中
        writeQueue.enqueue(SendResponseHeadersCommand.createTrailers(transportState(), http2Trailers, status), true);
    }
}

private void closeStreamWhenDone(ChannelPromise promise, int streamId) throws Http2Exception {
    final TransportState stream = this.serverStream(this.requireHttp2Stream(streamId));
    promise.addListener(new ChannelFutureListener() {
        public void operationComplete(ChannelFuture future) {
            stream.complete();
        }
    });
}

public void complete() {
    // 如果解帧器已经关闭了，则关闭监听器
    if (deframerClosed) {
        deframerClosedTask = null;
        closeListener(Status.OK);
    } else {
        // 如果还未关闭，则创建关闭监听器任务，并立即关闭解帧器
        deframerClosedTask = new Runnable() {
            @Override
            public void run() {
                closeListener(Status.OK);
            }
        };
        immediateCloseRequested = true;
        closeDeframer(true);
    }
}

private void closedInternal(final Status status) {
    // 如果状态不是 OK，则直接提交关闭 Context 任务
    if (!status.isOk()) {
        cancelExecutor.execute(new ContextCloser(context, status.getCause()));
    }
    final class Closed extends ContextRunnable {
        Closed() {
            super(context);
        }

        @Override
        public void runInContext() {
            PerfMark.startTask("ServerCallListener(app).closed", tag);
            PerfMark.linkIn(link);
            try {
                // 调用监听器的关闭事件
                getListener().closed(status);
            } finally {
                PerfMark.stopTask("ServerCallListener(app).closed", tag);
            }
        }
    }

    callExecutor.execute(new Closed());
}

final class Closed extends ContextRunnable {
    @Override
    public void runInContext() {
        try {
            // 调用监听器的关闭事件
            getListener().closed(status);
        } 
    }
}

private void closedInternal(Status status) {
    try {
        // 如果状态是 OK，通知监听器完成
        if (status.isOk()) {
            listener.onComplete();
        } else {
            // 否则将状态改为取消，通知监听器取消
            call.cancelled = true;
            listener.onCancel();
        }
    } finally {
        // 取消上下文
        context.cancel(null);
    }
}

public boolean cancel(Throwable cause) {
    boolean triggeredCancel = false;
    synchronized (this) {
        // 如果没有取消，则取消，并修改状态
        if (!cancelled) {
            cancelled = true;
            // 如果有等待取消的任务，则取消
            if (pendingDeadline != null) {
                pendingDeadline.cancel(false);
                pendingDeadline = null;
            }
            this.cancellationCause = cause;
            triggeredCancel = true;
        }
    }
    // 如果取消成功了，则通知监听器
    if (triggeredCancel) {
        notifyAndClearListeners();
    }
    return triggeredCancel;
}