gRPC学习记录(五)--拦截器分析
标签（空格分隔）： javaWEB
对于此类调用拦截器是必不可少的,本篇就分析下拦截器的实现.(博主本来想分析源码的,但是水平不足,并发知识欠缺,看的不是很懂,哎,仍需努力),另外貌似不同版本有些差异,这里使用的是1.0.3版本.
1.一个拦截器的小例子
在分析之前先看一种设计.
有一个接口如下:
/**
* 主调用接口
*/
public abstract class Client {
public abstract void start(String say);
}
/**
* 上述接口实现类
*/
public class ClientImp extends Client {
@Override
public void start(String say) {
System.out.println(say);
}
}
对此接口相关的转换器:
/**
* 用于包装Client到另一个Client
*/
public abstract class ForwardingClient extends Client{
//要包装的对象
protected abstract Client delegate();
@Override
public void start(String say) {
delegate().start(say);
}
}
/**
* 一个简单的包装实现类,必须要传入要包装的对象
*/
public class ForwardingClientImpl extends ForwardingClient{
//被委托对象
private final Client client;
public ForwardingClientImpl(Client client) {
this.client = client;
}
@Override
protected Client delegate() {
return client;
}
}
然后在下列方法中调用:
public class InterceptTest {
public static void main(String[] args) {
Client client = new ClientImp();//主要想执行的方法
//构造第一个拦截器
Client intercept1 = new ForwardingClientImpl(client){
@Override
public void start(String say) {
System.out.println("拦截器1");
super.start(say);
}
};
//构造第二个拦截器
Client intercept2 = new ForwardingClientImpl(intercept1){
@Override
public void start(String say) {
System.out.println("拦截器2");
super.start(say);
}
};
//执行主方法
intercept2.start("这是要执行的方法");
}
}
毫无疑问会输出
拦截器2
拦截器1
这是要执行的方法
分析一下针对Client接口,通过ForwardingClient可以实现自身的嵌套调用,从而达到了类似拦截器的效果.在gRPC中有很多类似的嵌套类,其本质和上面差不多,上面例子有助于对gRPC拦截器的掌握.
2.gRPC的ClientCall
该抽象类就是用来调用远程方法的,实现了发送消息和接收消息的功能,该接口由两个泛型ReqT和ReqT,分别对应着请求发送的信息,和请求收到的回复.
ClientCall抽象类主要有两个部分组成,一是public abstract static class Listener<T>用于监听服务端回复的消息,另一部分是针对客户端请求调用的一系列过程,如下代码流程所示:
该类中方法都是抽象方法,规定了整个调用顺序,如下:
call = channel.newCall(unaryMethod, callOptions);
call.start(listener, headers);
call.sendMessage(message);
call.halfClose();
call.request(1);
// wait for listener.onMessage()
在ClientCall的子类中有ForwardingClientCall<ReqT, RespT>,该类的作用和之前的Demo一样,用于包装ClientCall,然后实现委托嵌套调用,里面方法都如下代码所示:
@Override
public void start(Listener<RespT> responseListener, Metadata headers) {
delegate().start(responseListener, headers);
}
@Override
public void request(int numMessages) {
delegate().request(numMessages);
}
那和之前的Demo一对比,拦截器怎么使用就变得很容易了.
创建一个客户端拦截器,其中为header添加了token参数.之所以要实现ClientInterceptor接口,因为Channel本身也是可以嵌套的类,所以创建ClientCall也是被一层一层的调用.
/**
* 客户端拦截器
* @author Niu Li
* @date 2017/2/4
*/
//ClientInterceptor接口是针对ClientCall的创建进行拦截
public class ClientInterruptImpl implements ClientInterceptor {
@Override
public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(MethodDescriptor<ReqT, RespT> method,
CallOptions callOptions, Channel next) {
//创建client
System.out.println("创建client1");
ClientCall<ReqT,RespT> clientCall = next.newCall(method,callOptions);
return new ForwardingClientCall<ReqT, RespT>() {
@Override
protected ClientCall<ReqT, RespT> delegate() {
return clientCall;
}
@Override
public void start(Listener<RespT> responseListener, Metadata headers) {
System.out.println("拦截器1,在此可以对header参数进行修改");
Metadata.Key<String> token = Metadata.Key.of("token",Metadata.ASCII_STRING_MARSHALLER);
headers.put(token,"123456");
super.start(responseListener, headers);
}
};
}
}
调用输出如下:
创建client1
拦截器1,在此可以对header参数进行修改
这是针对客户端调用前的拦截,对于客户端收到的回复拦截则通过ClientCall的静态内部类Listener来实现,该Listener也是可以嵌套的,其内有如下方法:
public void onHeaders(Metadata headers) {}
public void onMessage(T message) {}
public void onClose(Status status, Metadata trailers) {}
public void onReady() {}
对之前start方法改造下,让其判断返回的header中有没有传送过去的token,没有则该请求视为失败.
@Override
public void start(Listener<RespT> responseListener, Metadata headers) {
System.out.println("拦截器1,在此可以对header参数进行修改");
Metadata.Key<String> token = Metadata.Key.of("token",Metadata.ASCII_STRING_MARSHALLER);
headers.put(token,"123456");
Listener<RespT> forwardListener = new ForwardingClientCallListener.
SimpleForwardingClientCallListener<RespT>(responseListener) {
@Override
public void onHeaders(Metadata headers) {
Metadata.Key<String> token = Metadata.Key.of("token",Metadata.ASCII_STRING_MARSHALLER);
if (!"123456".equals(headers.get(token))){
System.out.println("返回参数无token,关闭该链接");
super.onClose(Status.DATA_LOSS,headers);
}
super.onHeaders(headers);
}
};
super.start(forwardListener, headers);
}
最后再Channel创建的时候使用intercept(new ClientInterruptImpl())加入拦截器这样就简单实现了客户端的拦截了.
3.gRPC的ServerCall
有一点要搞明白,ClientCall是针对客户端要调用的方法的,而ServerCall是针对ClientCall的.看如下例子:
public class ServerInterruptImpl implements ServerInterceptor{
@Override
public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(ServerCall<ReqT, RespT> call,
Metadata headers, ServerCallHandler<ReqT, RespT> next) {
System.out.println("执行server拦截器1,获取token");
//获取客户端参数
Metadata.Key<String> token = Metadata.Key.of("token", Metadata.ASCII_STRING_MARSHALLER);
String tokenStr = headers.get(token);
if (StringUtil.isNullOrEmpty(tokenStr)){
System.out.println("未收到客户端token,关闭此连接");
call.close(Status.DATA_LOSS,headers);
}
//服务端写回参数
ServerCall<ReqT, RespT> serverCall = new ForwardingServerCall.SimpleForwardingServerCall<ReqT, RespT>(call) {
@Override
public void sendHeaders(Metadata headers) {
System.out.println("执行server拦截器2,写入token");
headers.put(token,tokenStr);
super.sendHeaders(headers);
}
};
return next.startCall(serverCall,headers);
}
}
当服务端接收到请求的时候就会打印出来如下的日志.这样就实现了服务端接手前的拦截和写回时的拦截.
执行server拦截器1,获取token
收到的信息:world:0
执行server拦截器2,写入token
关于更多使用还在琢磨中,目前欠缺并发知识,所以下一步打算看看并发相关的资料.
附录:
