gRPC Server 端启动流程

Server server = ServerBuilder.forPort(1234) 
                             .addService(new HelloServiceImpl()) 
                             .intercept(new CustomServerInterceptor())
                             .build();
server.start();
server.awaitTermination();

public static ServerBuilder<?> forPort(int port) {
    return ServerProvider.provider().builderForPort(port);
}

protected NettyServerBuilder builderForPort(int port) {
  return NettyServerBuilder.forPort(port);
}

private NettyServerBuilder(int port) {
    // 将本地 IP 和端口的地址添加到监听的地址集合中
    this.listenAddresses.add(new InetSocketAddress(port));
}

public final T addService(BindableService bindableService) {
    return addService(checkNotNull(bindableService, "bindableService").bindService());
}

public final io.grpc.ServerServiceDefinition bindService() {
    return io.grpc.ServerServiceDefinition.builder(getServiceDescriptor())
                                          .addMethod(getHowAreYouMethod(), asyncUnaryCall(new MethodHandlers<io.github.helloworlde.HelloMessage, io.github.helloworlde.HelloResponse>(this, METHODID_HOW_ARE_YOU)))
                                          .build();
}

MethodHandlers(HelloServiceImplBase serviceImpl, int methodId) {
    this.serviceImpl = serviceImpl;
    this.methodId = methodId;
}

@java.lang.Override
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 static <ReqT, RespT> ServerCallHandler<ReqT, RespT> asyncUnaryCall(UnaryMethod<ReqT, RespT> method) {
    return new UnaryServerCallHandler<>(method);
}

public final T addService(ServerServiceDefinition service) {
    // 将服务添加到处理器中
    registryBuilder.addService(checkNotNull(service, "service"));
    return thisT();
}

Builder addService(ServerServiceDefinition service) {
    services.put(service.getServiceDescriptor().getName(), service);
    return this;
}

public final T intercept(ServerInterceptor interceptor) {
    interceptors.add(checkNotNull(interceptor, "interceptor"));
    return thisT();
}

public final Server build() {
    return new ServerImpl(this, buildTransportServers(getTracerFactories()), Context.ROOT);
}

protected List<NettyServer> buildTransportServers(List<? extends ServerStreamTracer.Factory> streamTracerFactories) {
    assertEventLoopsAndChannelType();

    ProtocolNegotiator negotiator = protocolNegotiator;
    if (negotiator == null) {
        negotiator = sslContext != null ? ProtocolNegotiators.serverTls(sslContext, this.getExecutorPool()) : ProtocolNegotiators.serverPlaintext();
    }

    List<NettyServer> transportServers = new ArrayList<>(listenAddresses.size());
    // 为每一个监听的地址构建一个 NettyServer
    for (SocketAddress listenAddress : listenAddresses) {
        NettyServer transportServer = new NettyServer(listenAddress,
                channelFactory,
                channelOptions,
                childChannelOptions,
                bossEventLoopGroupPool,
                workerEventLoopGroupPool,
                forceHeapBuffer,
                negotiator,
                streamTracerFactories,
                getTransportTracerFactory(),
                maxConcurrentCallsPerConnection,
                autoFlowControl,
                flowControlWindow,
                maxMessageSize,
                maxHeaderListSize,
                keepAliveTimeInNanos,
                keepAliveTimeoutInNanos,
                maxConnectionIdleInNanos,
                maxConnectionAgeInNanos,
                maxConnectionAgeGraceInNanos,
                permitKeepAliveWithoutCalls,
                permitKeepAliveTimeInNanos,
                getChannelz());
        transportServers.add(transportServer);
    }
    return Collections.unmodifiableList(transportServers);
}

ServerImpl(AbstractServerImplBuilder<?> builder,
           List<? extends InternalServer> transportServers,
           Context rootContext) {
    this.executorPool = Preconditions.checkNotNull(builder.executorPool, "executorPool");
    this.registry = Preconditions.checkNotNull(builder.registryBuilder.build(), "registryBuilder");
    this.fallbackRegistry = Preconditions.checkNotNull(builder.fallbackRegistry, "fallbackRegistry");
    Preconditions.checkNotNull(transportServers, "transportServers");
    Preconditions.checkArgument(!transportServers.isEmpty(), "no servers provided");
    this.transportServers = new ArrayList<>(transportServers);
    this.logId = InternalLogId.allocate("Server", String.valueOf(getListenSocketsIgnoringLifecycle()));
    // Fork from the passed in context so that it does not propagate cancellation, it only
    // inherits values.
    this.rootContext = Preconditions.checkNotNull(rootContext, "rootContext").fork();
    this.decompressorRegistry = builder.decompressorRegistry;
    this.compressorRegistry = builder.compressorRegistry;
    this.transportFilters = Collections.unmodifiableList(new ArrayList<>(builder.transportFilters));
    this.interceptors = builder.interceptors.toArray(new ServerInterceptor[builder.interceptors.size()]);
    this.handshakeTimeoutMillis = builder.handshakeTimeoutMillis;
    this.binlog = builder.binlog;
    this.channelz = builder.channelz;
    this.serverCallTracer = builder.callTracerFactory.create();
    this.ticker = checkNotNull(builder.ticker, "ticker");
    channelz.addServer(this);
}

public ServerImpl start() throws IOException {
    synchronized (lock) {
        checkState(!started, "Already started");
        checkState(!shutdown, "Shutting down");
        // Start and wait for any ports to actually be bound.
        // 创建服务监听器实例
        ServerListenerImpl listener = new ServerListenerImpl();
        // 遍历 Server 并启动
        for (InternalServer ts : transportServers) {
            ts.start(listener);
            activeTransportServers++;
        }
        executor = Preconditions.checkNotNull(executorPool.getObject(), "executor");
        started = true;
        return this;
    }
}

@Override
public void start(ServerListener serverListener) throws IOException {
  listener = checkNotNull(serverListener, "serverListener");

  // 用于启动 Netty ServerChannel
  ServerBootstrap b = new ServerBootstrap();
  b.group(bossGroup, workerGroup);

  b.childHandler(new ChannelInitializer<Channel>() {
    @Override
    public void initChannel(Channel ch) {

      ChannelPromise channelDone = ch.newPromise();

      // 构建基于 Netty 的 ServerTransport
      NettyServerTransport transport = new NettyServerTransport(ch,
              channelDone,
              protocolNegotiator,
              streamTracerFactories,
              transportTracerFactory.create(),
              maxStreamsPerConnection,
              autoFlowControl,
              flowControlWindow,
              maxMessageSize,
              maxHeaderListSize,
              keepAliveTimeInNanos,
              keepAliveTimeoutInNanos,
              maxConnectionIdleInNanos,
              maxConnectionAgeInNanos,
              maxConnectionAgeGraceInNanos,
              permitKeepAliveWithoutCalls,
              permitKeepAliveTimeInNanos);
      ServerTransportListener transportListener;
      synchronized (NettyServer.this) {
        // 调用监听器回调，Transport 创建事件
        transportListener = listener.transportCreated(transport);
      }

      // 启动监听器
      transport.start(transportListener);
    }
  });
  // 绑定地址，接受连接
  ChannelFuture future = b.bind(address);

  channel = future.channel();
  channel.eventLoop().execute(new Runnable() {
    @Override
    public void run() {
      listenSocketStats = new ListenSocket(channel);
      channelz.addListenSocket(listenSocketStats);
    }
  });
}

@Override
public ServerTransportListener transportCreated(ServerTransport transport) {
    synchronized (lock) {
        transports.add(transport);
    }
    // Transport 监听器
    ServerTransportListenerImpl stli = new ServerTransportListenerImpl(transport);
    // 初始化监听器，添加握手超时的取消任务
    stli.init();
    return stli;
}

public void start(ServerTransportListener listener) {
    Preconditions.checkState(this.listener == null, "Handler already registered");
    this.listener = listener;

    // Create the Netty handler for the pipeline.
    // 为 pipeline 创建 Netty Handler
    grpcHandler = createHandler(listener, channelUnused);

    // Notify when the channel closes.
    // 当 Channel 关闭的时候发送通知
    final class TerminationNotifier implements ChannelFutureListener {
        boolean done;

        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
            if (!done) {
                done = true;
                notifyTerminated(grpcHandler.connectionError());
            }
        }
    }

    // 创建 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 awaitTermination() throws InterruptedException {
    synchronized (lock) {
        while (!terminated) {
            lock.wait();
        }
    }
}