网站建设來超速云建站,精准引流推广,wordpress视频模版,手机网站开发需求文档前言
Dubbo 协议层的核心SPI接口是org.apache.dubbo.rpc.Protocol#xff0c;通过扩展该接口和围绕的相关接口#xff0c;就可以让 Dubbo 使用我们自定义的协议来通信。默认的协议是 dubbo#xff0c;本文提供一个 Grpc 协议的实现。
设计思路 Google 提供了 Java 的 Grpc…前言
Dubbo 协议层的核心SPI接口是org.apache.dubbo.rpc.Protocol通过扩展该接口和围绕的相关接口就可以让 Dubbo 使用我们自定义的协议来通信。默认的协议是 dubbo本文提供一个 Grpc 协议的实现。
设计思路 Google 提供了 Java 的 Grpc 实现所以我们站在巨人的肩膀上即可就不用重复造轮子了。 首先我们要实现 Protocol 接口服务暴露时开启我们的 GrpcServer绑定本地端口用于后续处理连接和请求。 服务端如何处理grpc请求呢 方案一是把暴露的所有服务 Invoker 都封装成grpc的 Service全部统一让 GrpcServer 处理但是这么做太麻烦了。方案二是提供一个 DispatcherService统一处理客户端发来的grpc请求再根据参数查找要调用的服务执行本地调用返回结果。本文采用方案二。 客户端引用服务时我们创建 GrpcInvoker 对象和服务端建立连接并生成 DispatcherService 的本地存根 Stub 对象发起 RPC 调用时只需把 RpcInvocation 转换成 Protobuf 消息发出去即可。
实现GrpcProtocol
项目结构
首先我们新建一个dubbo-extension-protocol-grpc模块引入必要的依赖。
dependenciesdependencygroupIdorg.apache.dubbo/groupIdartifactIddubbo-rpc-api/artifactIdversion${dubbo.version}/version/dependencydependencygroupIdio.grpc/groupIdartifactIdgrpc-all/artifactIdversion1.56.1/version/dependency
/dependencies项目结构
main
--java
----dubbo.extension.rpc.grpc
------message
--------RequestData.java
--------ResponseData.java
------Codec.java
------DispatcherService.java
------DispatcherServiceGrpc.java
------GrpcExporter.java
------GrpcInvoker.java
------GrpcProtocol.java
------GrpcProtocolServer.java
--resources
----META-INF/dubbo
------org.apache.dubbo.rpc.Protocol服务消息定义
然后是定义grpc的 Service 和消息格式 DispatcherService.proto 请求分发服务的定义
syntax proto3;option java_multiple_files true;
option java_package dubbo.extension.rpc.grpc;
option java_outer_classname DispatcherServiceProto;
option objc_class_prefix HLW;import RequestData.proto;
import ResponseData.proto;service DispatcherService {rpc dispatch (RequestData) returns (ResponseData) {}
}RequestData.proto 请求消息的定义主要是对 Invocation 的描述
syntax proto3;option java_multiple_files true;
option java_package dubbo.extension.rpc.grpc.message;
option java_outer_classname RequestDataProto;
option objc_class_prefix HLW;message RequestData {string targetServiceUniqueName 1;string methodName 2;string serviceName 3;repeated bytes parameterTypes 4;string parameterTypesDesc 5;repeated bytes arguments 6;bytes attachments 7;
}ResponseData.proto 响应消息的定义主要是对 AppResponse 的描述
syntax proto3;option java_multiple_files true;
option java_package dubbo.extension.rpc.grpc.message;
option java_outer_classname ResponseataProto;
option objc_class_prefix HLW;message ResponseData {int32 status 1;string errorMessage 2;bytes result 3;bytes attachments 4;
}使用protobuf-maven-plugin插件把 proto 文件生成对应的 Java 类。
协议实现
新建 GrpcProtocol 类继承 AbstractProtocol实现 Protocol 协议细节。 核心是服务暴露时开启 Grpc 服务引用服务时生成对应的 Invoker。
public class GrpcProtocol extends AbstractProtocol {Overrideprotected T InvokerT protocolBindingRefer(ClassT type, URL url) throws RpcException {return new GrpcInvoker(type, url);}Overridepublic int getDefaultPort() {return 18080;}Overridepublic T ExporterT export(InvokerT invoker) throws RpcException {GrpcExporterT exporter new GrpcExporter(invoker);exporterMap.put(invoker.getInterface().getName(), exporter);openServer(invoker.getUrl());return exporter;}private void openServer(URL url) {String key serviceKey(url);ProtocolServer protocolServer serverMap.get(key);if (protocolServer null) {synchronized (serverMap) {protocolServer serverMap.get(key);if (protocolServer null) {serverMap.put(key, createServer(url));}}}}private ProtocolServer createServer(URL url) {return new GrpcProtocolServer(url, exporterMap);}
}新建 GrpcProtocolServer 类实现 ProtocolServer 接口核心是启动 GrpcServer并添加 DispatcherService 处理请求。
public class GrpcProtocolServer implements ProtocolServer {private final Server server;public GrpcProtocolServer(URL url, MapString, Exporter? exporterMap) {server ServerBuilder.forPort(url.getPort()).addService(new DispatcherService(exporterMap)).build();try {server.start();} catch (IOException e) {throw new RuntimeException(e);}}Overridepublic String getAddress() {return null;}Overridepublic void setAddress(String address) {}Overridepublic void close() {server.shutdown();}
}新建 DispatcherService 类实现 Grpc Service用来处理客户端的grpc请求。核心是把 RequestData 解码成 RpcInvocation再查找本地 Invoker 调用并返回结果。
public class DispatcherService extends DispatcherServiceGrpc.DispatcherServiceImplBase {private final MapString, Exporter? exporterMap;public DispatcherService(MapString, Exporter? exporterMap) {this.exporterMap exporterMap;}Overridepublic void dispatch(RequestData request, StreamObserverResponseData responseObserver) {RpcInvocation invocation Codec.decodeInvocation(request);ResponseData responseData;try {Invoker? invoker exporterMap.get(invocation.getServiceName()).getInvoker();Object returnValue invoker.invoke(invocation).get().getValue();responseData Codec.encodeResponse(returnValue, null);} catch (Exception e) {responseData Codec.encodeResponse(null, e);}responseObserver.onNext(responseData);responseObserver.onCompleted();}
}新建 GrpcInvoker 类实现 Invoker 接口服务引用时会创建它目的是发起 RPC 调用时通过 Stub 发一个请求到 DispatcherService实现grpc协议的 RPC 调用。
public class GrpcInvokerT extends AbstractInvokerT {private static final MapString, DispatcherServiceGrpc.DispatcherServiceFutureStub STUB_MAP new ConcurrentHashMap();public GrpcInvoker(ClassT type, URL url) {super(type, url);}private DispatcherServiceGrpc.DispatcherServiceFutureStub getStub() {String key getUrl().getAddress();DispatcherServiceGrpc.DispatcherServiceFutureStub stub STUB_MAP.get(key);if (stub null) {synchronized (STUB_MAP) {stub STUB_MAP.get(key);if (stub null) {STUB_MAP.put(key, stub createClient(getUrl()));}}}return stub;}private DispatcherServiceGrpc.DispatcherServiceFutureStub createClient(URL url) {ManagedChannel channel ManagedChannelBuilder.forAddress(url.getHost(), url.getPort()).usePlaintext().build();return DispatcherServiceGrpc.newFutureStub(channel);}Overrideprotected Result doInvoke(Invocation invocation) throws Throwable {RequestData requestData Codec.encodeInvocation((RpcInvocation) invocation);ResponseData responseData getStub().dispatch(requestData).get();return Codec.decodeResponse(responseData, invocation);}
}最后是编解码器 Codec它的作用是对 RequestData、ResponseData 对象的编解码。对于请求来说要编解码的是 RpcInvocation对于响应来说要编解码的是返回值和异常信息。 方法实参是 Object[] 类型附带参数是 Map 类型本身不能直接通过 Protobuf 传输我们会先利用 Serialization 序列化成字节数组后再传输。
public class Codec {private static final Serialization serialization ExtensionLoader.getExtensionLoader(Serialization.class).getDefaultExtension();public static RequestData encodeInvocation(RpcInvocation invocation) {RequestData.Builder builder RequestData.newBuilder().setTargetServiceUniqueName(invocation.getTargetServiceUniqueName()).setMethodName(invocation.getMethodName()).setServiceName(invocation.getServiceName());for (Class? parameterType : invocation.getParameterTypes()) {builder.addParameterTypes(serialize(parameterType));}builder.setParameterTypesDesc(invocation.getParameterTypesDesc());for (Object argument : invocation.getArguments()) {builder.addArguments(serialize(argument));}builder.setAttachments(serialize(invocation.getAttachments()));return builder.build();}public static RpcInvocation decodeInvocation(RequestData requestData) {RpcInvocation invocation new RpcInvocation();invocation.setTargetServiceUniqueName(requestData.getTargetServiceUniqueName());invocation.setMethodName(requestData.getMethodName());invocation.setServiceName(requestData.getServiceName());ListByteString parameterTypesList requestData.getParameterTypesList();Class?[] parameterTypes new Class[parameterTypesList.size()];for (int i 0; i parameterTypesList.size(); i) {parameterTypes[i] (Class?) deserialize(parameterTypesList.get(i));}invocation.setParameterTypes(parameterTypes);invocation.setParameterTypesDesc(requestData.getParameterTypesDesc());ListByteString argumentsList requestData.getArgumentsList();Object[] arguments new Object[argumentsList.size()];for (int i 0; i argumentsList.size(); i) {arguments[i] deserialize(argumentsList.get(i));}invocation.setArguments(arguments);invocation.setAttachments((MapString, String) deserialize(requestData.getAttachments()));return invocation;}public static Result decodeResponse(ResponseData responseData, Invocation invocation) {AppResponse appResponse new AppResponse();if (responseData.getStatus() 200) {appResponse.setValue(deserialize(responseData.getResult()));appResponse.setAttachments((MapString, String) deserialize(responseData.getAttachments()));} else {appResponse.setException(new RuntimeException(responseData.getErrorMessage()));}return new AsyncRpcResult(CompletableFuture.completedFuture(appResponse), invocation);}private static Object deserialize(ByteString byteString) {try {InputStream inputStream new ByteArrayInputStream(byteString.toByteArray());ObjectInput objectInput serialization.deserialize(null, inputStream);return objectInput.readObject();} catch (Exception e) {throw new RuntimeException(e);}}private static ByteString serialize(Object obj) {try {ByteArrayOutputStream outputStream new ByteArrayOutputStream();ObjectOutput output serialization.serialize(null, outputStream);output.writeObject(obj);output.flushBuffer();return ByteString.copyFrom(outputStream.toByteArray());} catch (Exception e) {throw new RuntimeException(e);}}public static ResponseData encodeResponse(Object returnValue, Throwable throwable) {ResponseData.Builder builder ResponseData.newBuilder();if (throwable null) {builder.setStatus(200);builder.setResult(serialize(returnValue));builder.setAttachments(serialize(new HashMap()));//先忽略} else {builder.setStatus(500);builder.setErrorMessage(throwable.getMessage());}return builder.build();}
}实现完毕最后是让 Dubbo 可以加载到我们自定义的 GrpcProtocol可以通过 SPI 的方式。新建META-INF/dubbo/org.apache.dubbo.rpc.Protocol文件内容
grpcdubbo.extension.rpc.grpc.GrpcProtocol服务提供方使用自定义协议
ProtocolConfig protocolConfig new ProtocolConfig(grpc, 10880);消费方使用自定义协议
ReferenceConfig#setUrl(grpc://127.0.0.1:10880);尾巴
Protocol 层关心的是如何暴露服务和引用服务以及如何让双方使用某个具体的协议来通信以完成 RPC 调用。如果你觉得官方提供的 dubbo 协议无法满足你的业务就可以通过扩展 Protocol 接口来实现你自己的私有协议。
