基于grpc从零开始搭建一个准生产分布式应用(4) - 04 - grpc框架级应用

开始前必读：??基于grpc从零开始搭建一个准生产分布式应用(0) - quickStart???

原生GRPC使用的最后一个章节，这里只描述一些重要的经常用到的内容。主要用于在集成grpc时的一些底层设置。

一、拦截器

下面的例子只是其中一种写法，可以研究下源码。这块是有一套继承体系的。

1.1、客户端拦截

public class OrderClientInterceptor implements ClientInterceptor {
    @Override
    public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel channel) {

        return new ForwardingClientCall.SimpleForwardingClientCall<ReqT, RespT>(channel.newCall(method, callOptions)) {
            @Override
            public void start(Listener<RespT> responseListener, Metadata headers) {
                headers.put(Metadata.Key.of("MY_MD_1", ASCII_STRING_MARSHALLER), "This is metadata of MY_MD_1");
                super.start(responseListener, headers);
            }
        };
    }
}

public class OrderClientInterceptor implements ClientInterceptor {
    @Override
    public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
        return null;
    }
}

1.2、服务端拦截

public class OrderMgtServerInterceptor implements io.grpc.ServerInterceptor {
    private static final Logger logger = Logger.getLogger(OrderMgtServerInterceptor.class.getName());

    @Override
    public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall
    (ServerCall<ReqT, RespT> call, Metadata headers, ServerCallHandler<ReqT, RespT> next) {

        logger.info("======= [Server Interceptor] : Remote Method Invoked - " + call.getMethodDescriptor().getFullMethodName());
        ServerCall<ReqT, RespT> serverCall = new OrderMgtServerCall<>(call);
        return new OrderMgtServerCallListener<>(next.startCall(serverCall, headers));
    }

}

public class OrderMgtServerCallListener<R> extends ForwardingServerCallListener<R> {
    private static final Logger logger = Logger.getLogger(OrderMgtServerCallListener.class.getName());

    private final ServerCall.Listener<R> delegate;

    OrderMgtServerCallListener(ServerCall.Listener<R> delegate) {
        this.delegate = delegate;
    }

    @Override
    protected ServerCall.Listener<R> delegate() {
        return delegate;
    }

    @Override
    public void onMessage(R message) {
        logger.info("Message Received from Client -> Service " + message);
        super.onMessage(message);
    }
}

public class OrderMgtServerCall<ReqT, RespT> extends ForwardingServerCall.SimpleForwardingServerCall<ReqT, RespT>  {

    private static final Logger logger = Logger.getLogger(OrderMgtServerCall.class.getName());

    OrderMgtServerCall(ServerCall<ReqT, RespT> delegate) {
        super(delegate);
    }

    @Override
    protected ServerCall<ReqT, RespT> delegate() {
        return super.delegate();
    }

    @Override
    public MethodDescriptor<ReqT, RespT> getMethodDescriptor() {
        return super.getMethodDescriptor();
    }

    @Override
    public void sendMessage(RespT message) {
        logger.info("Message from Service -> Client : " + message);
        super.sendMessage(message);
    }
}

二、截止时间

一个请求可能会调用多个RPC，每个RPC都有超时时间。但这个超时时间不能做为服务的整个生命周期，这时需要使用截止时间。如果客户端没有设置这个时间，就会无限期的等待下去，这个会有耗尽资源的风险。

在grpc中没有超时时间。在客户端到达指定的截止时间后，服务端还是有可能会做出响应的。但服务端也可以随时探测到客户端的情况。这个功能需要在客户端设置，如果想在服务端设置的话：可以采用线程池把每每个事务访问缓存然后轮询过期时间的方式间接实现这个超时功能，最好用异步的方式实现。

public static void main(String[] args) {
    ManagedChannel channel = ManagedChannelBuilder.forAddress(
            "localhost", 50051).usePlaintext().build();
    //设置截止时间
    OrderManagementGrpc.OrderManagementBlockingStub stub =
    OrderManagementGrpc.newBlockingStub(channel).withDeadlineAfter(1000, TimeUnit.MILLISECONDS);
    OrderManagementGrpc.OrderManagementStub asyncStub = OrderManagementGrpc.newStub(channel);

    OrderManagementOuterClass.Order order = OrderManagementOuterClass.Order
            .newBuilder()
            .setId("101")
            .addItems("iPhone XS").addItems("Mac Book Pro")
            .setDestination("San Jose, CA")
            .setPrice(2300)
            .build();


    try {
        // Add Order with a deadline
        StringValue result = stub.addOrder(order);
        logger.info("AddOrder Response -> : " + result.getValue());
    } catch (StatusRuntimeException e) {
        //捕获异常
        if (e.getStatus().getCode() == Status.Code.DEADLINE_EXCEEDED) {
            logger.info("Deadline Exceeded. : " + e.getMessage());
        } else {
            logger.info("Unspecified error from the service -> " + e.getMessage());
        }
    }

}

三、错误处理

其实就是异常捕获。示例代码如下：

3.1、服务端

如果用trycatch-finally结构的话，注意finally中不能放onCompleted()方法，原因是onError()和onCompleted()会调用同一段状态代码，导致二次关闭。所以在finally不是太必要使用或是放置一些Log代码也可。

public void addOrder(OrderManagementOuterClass.Order request, StreamObserver<StringValue> responseObserver) {

    // Handling invalid orders
    if (request.getId().equals("-1")) {
        logger.info("Invalid Order ID: " + request.getId());
        //Status这个类里有很多的错误码可用
        responseObserver.onError(Status.INVALID_ARGUMENT.withDescription("Invalid order ID received.").asException());
    }

    orderMap.put(request.getId(), request);
    StringValue id = StringValue.newBuilder().setValue("100500").build();
    responseObserver.onNext(id);
    responseObserver.onCompleted();
}

3.2、客户端

try {
    // Add Order with a deadline
    StringValue result = stub.addOrder(order);
    logger.info("AddOrder Response -> : " + result.getValue());
} catch (StatusRuntimeException e) {
    logger.info(" Error Received - Error Code : " + e.getStatus().getCode());
    logger.info("Error details -> " + e.getMessage());
}

3.3、错误码

往往在开发时，每个小组都习惯性的定义一些错误码做为规范。从过往经验来看不太可取，原因有几个：1、需开发一个base包供所有应用集成，而且又不太可能一次把所有异常码全定义完整，面临升级问题；2、不同部门间交互问题，可以说其它部门的人很少能遵守另一部门的内部规范。所以如果能用开源框架提供的定义就使用或是采用一些事实上的标准比如http200,400等。下面是GRPC提供的所有错误码：

/**io.grpc.Status.java详细定义如下*/
0：成功  - 建议使用
1：操作被调用者取消
2：未知错误  - 建议使用
3：客户端指定了非法参数  - 建议使用
4：服务端处理超时
5：某些请求实体未找到
6：客户端试图创建的实体已存在
7：调用者没有权限执行特定的操作
8：某些资源被耗尽
9：操作被拒绝
10：操作被中止
11：尝试进行的操作超出了合法的范围  - 建议使用
12：在此服务中，未实现或不支持本操作  - 建议使用
13：内部错误  - 建议使用
14：此服务当前不可用
15：不可恢复的数据丢失或损坏
16：客户端没有进行操作的合法认证凭证

四、元数据

其实就是线程数据，同一线程可共享，也可用于隐匿传参。一般包含协议头数据，也可以人为通过拦截器设置应用数据。在GRPC是非ThreadLoacl来实现的，所以框架会自己清理，不必担心OOM问题。参考下面的拦截器例子中的headers.put()代码块。

public class OrderClientInterceptor implements ClientInterceptor {
    @Override
    public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel channel) {

        return new ForwardingClientCall.SimpleForwardingClientCall<ReqT, RespT>(channel.newCall(method, callOptions)) {
            @Override
            public void start(Listener<RespT> responseListener, Metadata headers) {
                headers.put(Metadata.Key.of("MY_MD_1", ASCII_STRING_MARSHALLER), "This is metadata of MY_MD_1");
                super.start(responseListener, headers);
            }
        };
    }
}

五、压缩

有时这个不建议使用，因为需要两方使用同一种压缩器，在服务器端已注册的压缩器会自动解压

StringValue result = stub.withCompression("gzip").addOrder(order);

OrderManagementOuterClass.Order orderResponse = stub.withCompression("gzip").getOrder(id);