自定义Dubbo RPC通信协议

前言

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模块，引入必要的依赖。

<dependencies><dependency><groupId>org.apache.dubbo</groupId><artifactId>dubbo-rpc-api</artifactId><version>${dubbo.version}</version></dependency><dependency><groupId>io.grpc</groupId><artifactId>grpc-all</artifactId><version>1.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> Invoker<T> protocolBindingRefer(Class<T> type, URL url) throws RpcException {return new GrpcInvoker<>(type, url);}@Overridepublic int getDefaultPort() {return 18080;}@Overridepublic <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {GrpcExporter<T> 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, Map<String, 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 Map<String, Exporter<?>> exporterMap;public DispatcherService(Map<String, Exporter<?>> exporterMap) {this.exporterMap = exporterMap;}@Overridepublic void dispatch(RequestData request, StreamObserver<ResponseData> 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 GrpcInvoker<T> extends AbstractInvoker<T> {private static final Map<String, DispatcherServiceGrpc.DispatcherServiceFutureStub> STUB_MAP = new ConcurrentHashMap<>();public GrpcInvoker(Class<T> 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());List<ByteString> 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());List<ByteString> 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((Map<String, 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((Map<String, 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文件，内容：

grpc=dubbo.extension.rpc.grpc.GrpcProtocol

服务提供方使用自定义协议：

ProtocolConfig protocolConfig = new ProtocolConfig("grpc", 10880);

消费方使用自定义协议：

ReferenceConfig#setUrl("grpc://127.0.0.1:10880");

尾巴

Protocol 层关心的是如何暴露服务和引用服务，以及如何让双方使用某个具体的协议来通信，以完成 RPC 调用。如果你觉得官方提供的 dubbo 协议无法满足你的业务，就可以通过扩展 Protocol 接口来实现你自己的私有协议。