《Java 服务端通信工程》4.1 Stub 与 Channel 生命周期

package com.plume.linkd.v3.core.client;

import com.plume.linkd.v3.api.Book;
import com.plume.linkd.v3.api.GetBookRequest;
import com.plume.linkd.v3.api.LibraryGrpc;
import io.grpc.InsecureChannelCredentials;
import io.grpc.ManagedChannel;
import io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder;
import java.util.concurrent.TimeUnit;

public final class LibraryClient implements AutoCloseable {

    private final ManagedChannel channel;
    private final LibraryGrpc.LibraryBlockingStub blockingStub;
    private final LibraryGrpc.LibraryStub asyncStub;

    public LibraryClient(String host, int port) {
        this.channel = NettyChannelBuilder
                .forAddress(host, port, InsecureChannelCredentials.create())
                .defaultLoadBalancingPolicy("round_robin")
                .maxInboundMessageSize(4 * 1024 * 1024)
                .build();
        this.blockingStub = LibraryGrpc.newBlockingStub(channel);
        this.asyncStub = LibraryGrpc.newStub(channel);
    }

    public Book getBook(String isbn) {
        return blockingStub
                .withDeadlineAfter(300, TimeUnit.MILLISECONDS)
                .getBook(GetBookRequest.newBuilder().setIsbn(isbn).build());
    }

    public LibraryGrpc.LibraryStub async() {
        return asyncStub;
    }

    @Override
    public void close() {
        channel.shutdown();
        try {
            if (!channel.awaitTermination(5, TimeUnit.SECONDS)) {
                channel.shutdownNow();
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            channel.shutdownNow();
        }
    }
}

package com.plume.linkd.v3.core.client;

import io.grpc.ConnectivityState;
import io.grpc.ManagedChannel;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

public final class ChannelWarmup {

    public static boolean awaitReady(ManagedChannel channel, long timeoutMillis)
            throws InterruptedException {
        long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(timeoutMillis);
        while (true) {
            ConnectivityState state = channel.getState(true);
            if (state == ConnectivityState.READY) {
                return true;
            }
            if (state == ConnectivityState.SHUTDOWN) {
                return false;
            }
            long remaining = deadline - System.nanoTime();
            if (remaining <= 0L) {
                return false;
            }
            CountDownLatch changed = new CountDownLatch(1);
            channel.notifyWhenStateChanged(state, changed::countDown);
            changed.await(remaining, TimeUnit.NANOSECONDS);
        }
    }
}

// 变体一：停机预算很短，直接取消
channel.shutdownNow();
channel.awaitTermination(1, TimeUnit.SECONDS);

// 变体二：先停止新调用，等在途请求自己结束
channel.shutdown();
if (!channel.awaitTermination(5, TimeUnit.SECONDS)) {
    channel.shutdownNow();
    channel.awaitTermination(1, TimeUnit.SECONDS);
}

LibraryClient library = new LibraryClient("libraryd.internal", 9093);
if (!ChannelWarmup.awaitReady(library.channel(), 1_000)) {
    LOG.log(System.Logger.Level.WARNING, "libraryd 未在 1s 内就绪，降级启动");
}
Runtime.getRuntime().addShutdownHook(new Thread(library::close, "linkd-grpc-close"));

package com.plume.linkd.v3.probe;

import io.grpc.CallOptions;
import io.grpc.InsecureChannelCredentials;
import io.grpc.ManagedChannel;
import io.grpc.MethodDescriptor;
import io.grpc.Server;
import io.grpc.ServerServiceDefinition;
import io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder;
import io.grpc.netty.shaded.io.grpc.netty.NettyServerBuilder;
import io.grpc.stub.ClientCalls;
import io.grpc.stub.ServerCalls;
import java.io.ByteArrayInputStream;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeUnit;

public final class UnaryProbe {

    static final MethodDescriptor.Marshaller<String> UTF8 = new MethodDescriptor.Marshaller<>() {
        @Override
        public InputStream stream(String value) {
            return new ByteArrayInputStream(value.getBytes(StandardCharsets.UTF_8));
        }

        @Override
        public String parse(InputStream stream) {
            try {
                return new String(stream.readAllBytes(), StandardCharsets.UTF_8);
            } catch (java.io.IOException e) {
                throw new IllegalStateException("read utf8", e);
            }
        }
    };

    public static void main(String[] args) throws Exception {
        MethodDescriptor<String, String> method = MethodDescriptor.<String, String>newBuilder()
                .setType(MethodDescriptor.MethodType.UNARY)
                .setFullMethodName("plume.linkd.v3.Library/GetBook")
                .setRequestMarshaller(UTF8)
                .setResponseMarshaller(UTF8)
                .build();

        Server server = NettyServerBuilder.forPort(0)
                .addService(ServerServiceDefinition.builder("plume.linkd.v3.Library")
                        .addMethod(method, ServerCalls.asyncUnaryCall((request, observer) -> {
                            observer.onNext("book:42");
                            observer.onCompleted();
                        }))
                        .build())
                .build()
                .start();

        ManagedChannel channel = NettyChannelBuilder
                .forAddress("127.0.0.1", server.getPort(), InsecureChannelCredentials.create())
                .build();
        try {
            String reply = ClientCalls.blockingUnaryCall(channel, method,
                    CallOptions.DEFAULT.withDeadlineAfter(800, TimeUnit.MILLISECONDS), "42");
            System.out.println("unary=" + reply);
        } finally {
            channel.shutdownNow().awaitTermination(1, TimeUnit.SECONDS);
            server.shutdownNow().awaitTermination(1, TimeUnit.SECONDS);
        }
    }
}

unary=book:42

WARNING: A terminally deprecated method in sun.misc.Unsafe has been called
WARNING: sun.misc.Unsafe::allocateMemory has been called by io.grpc.netty.shaded.io.netty.util.internal.PlatformDependent0$2
WARNING: Please consider reporting this to the maintainers of class io.grpc.netty.shaded.io.netty.util.internal.PlatformDependent0$2
WARNING: sun.misc.Unsafe::allocateMemory will be removed in a future release

// 反例：每个请求一条 channel
for (String isbn : isbns) {
    try (ManagedChannel ch = NettyChannelBuilder
            .forAddress("libraryd.internal", 9093, InsecureChannelCredentials.create())
            .build()) {
        LibraryGrpc.newBlockingStub(ch).getBook(request(isbn));
    }
}