前言
异步编程中，callback是经常用到的结果获取方式，做为一个负责任的框架，Dubbo自然也要提供对Callback的支持。相对于进程内的异步结果返回，RPC中对异步的支持更加复杂一些，在Consumer端和Provider都要提供相应的支持。
实现原理
Callback方法的声明
Dubbo中如果有个接口的方法是Callback方法，需要在url中声明第几个参数是Callback回调接口。如下接口：
public interface DemoService {
void sayHello(String to, Callback resp);
}
sayHello方法的第2参数用来接收返回结果，则url中需要添加参数sayHello.1.callback=true。
Dubbo对Callback的实现非常巧妙。当调用非Callback方法时，Provider端暴露服务，Consumer端生成一个代理，通过Invoker发送请求，Provider端收到请求回复一个Response。
当调用的是Callback方法时，Consumer端发送请求的同时暴露一个回调参数的服务，这样Provider返回结果的方式就变成了调用Consumer暴露的这个服务，也就是返回结果时Provider变成了Consumer。很好了复用了Dubbo本身的逻辑。
调用端支持
在Consumer端，用户调用异步方法和普通调用没有区别，代理层的Invoker也没有对callback方法做任何的特殊处理。对Callback方法的处理是在传输层中，在DubboCodec中对Request进行协议编码时，发现方法参数是callback参数，则需要暴露一个服务提供给Provider回调使用：
@Override
protected void encodeRequestData(Channel channel, ObjectOutput out, Object data, String version) throws IOException {
RpcInvocation inv = (RpcInvocation) data;
out.writeUTF(version);
out.writeUTF(inv.getAttachment(PATH_KEY));
out.writeUTF(inv.getAttachment(VERSION_KEY));
out.writeUTF(inv.getMethodName());
out.writeUTF(inv.getParameterTypesDesc());
Object[] args = inv.getArguments();
if (args != null) {
for (int i = 0; i < args.length; i++) {
//写入调用参数
out.writeObject(encodeInvocationArgument(channel, inv, i));
}
}
out.writeAttachments(inv.getObjectAttachments());
}
public static Object encodeInvocationArgument(Channel channel, RpcInvocation inv, int paraIndex) throws IOException {
// get URL directly
URL url = inv.getInvoker() == null ? null : inv.getInvoker().getUrl();
//判断是否为Callback方法
byte callbackStatus = isCallBack(url, inv.getMethodName(), paraIndex);
Object[] args = inv.getArguments();
Class<?>[] pts = inv.getParameterTypes();
switch (callbackStatus) {
//如果是Callback方法，则暴露服务并且将instanceid放入attachment中带到Server端
case CallbackServiceCodec.CALLBACK_CREATE:
inv.setAttachment(INV_ATT_CALLBACK_KEY + paraIndex, exportOrUnexportCallbackService(channel, url, pts[paraIndex], args[paraIndex], true));
return null;
case CallbackServiceCodec.CALLBACK_DESTROY:
inv.setAttachment(INV_ATT_CALLBACK_KEY + paraIndex, exportOrUnexportCallbackService(channel, url, pts[paraIndex], args[paraIndex], false));
return null;
default:
return args[paraIndex];
}
}
上面的逻辑中，DubboCodec在对调用参数进行编码时，判断出是回调方法，则将调用exportOrUnexportCallbackService()将传入的参数暴露成一个服务。
private static String exportOrUnexportCallbackService(Channel channel, URL url, Class clazz, Object inst, Boolean export) throws IOException {
//为回调参数生成一个唯一的id，使用的jvm生成的hashcode
int instid = System.identityHashCode(inst);
Map<String, String> params = new HashMap<>(3);
// url中添加参数isserver=false，代表这是client暴露的服务
params.put(IS_SERVER_KEY, Boolean.FALSE.toString());
// url中添加参数is_callback_service=true
params.put(IS_CALLBACK_SERVICE, Boolean.TRUE.toString());
String group = (url == null ? null : url.getParameter(GROUP_KEY));
if (group != null && group.length() > 0) {
params.put(GROUP_KEY, group);
}
//添加方法声明
params.put(METHODS_KEY, StringUtils.join(Wrapper.getWrapper(clazz).getDeclaredMethodNames(), ","));
Map<String, String> tmpMap = new HashMap<>(url.getParameters());
tmpMap.putAll(params);
tmpMap.remove(VERSION_KEY);// doesn't need to distinguish version for callback
tmpMap.put(INTERFACE_KEY, clazz.getName());
//生成callback服务的url
URL exportUrl = new URL(DubboProtocol.NAME, channel.getLocalAddress().getAddress().getHostAddress(), channel.getLocalAddress().getPort(), clazz.getName() + "." + instid, tmpMap);
String cacheKey = getClientSideCallbackServiceCacheKey(instid);
String countKey = getClientSideCountKey(clazz.getName());
if (export) {
// 对于同一个instance只会暴露一次
if (!channel.hasAttribute(cacheKey)) {
//限制对于同一个callback，限制并发数
if (!isInstancesOverLimit(channel, url, clazz.getName(), instid, false)) {
//生成Invoker
Invoker<?> invoker = PROXY_FACTORY.getInvoker(inst, clazz, exportUrl);
// export callback服务
Exporter<?> exporter = protocol.export(invoker);
channel.setAttribute(cacheKey, exporter);
}
}
} else {
//销毁callback服务
if (channel.hasAttribute(cacheKey)) {
Exporter<?> exporter = (Exporter<?>) channel.getAttribute(cacheKey);
exporter.unexport();
channel.removeAttribute(cacheKey);
decreaseInstanceCount(channel, countKey);
}
}
return String.valueOf(instid);
}
服务暴露也是调用的DubboProtocol.export()，跟普通的服务保留没有区别，只是会在url中额外添加参数标识是callback服务。有一点跟普通的服务暴露有点区别就是，不会额外开启Server，而是使用发送请求的channel。
服务提供方支持
服务提供方在收到callback接口的请求后，也是在进入DubboCodec进行解码的时候识别出是Callback方法，就会生成参数的代理来调用Provider端的本地方法。本地方法在处理完，调用Callback参数返回结果时，实际是调用Proxy，进而发出远程调用。
首先看下DubboCodec中的decode过程：
@Override
protected Object decodeBody(Channel channel, InputStream is, byte[] header) throws IOException {
byte flag = header[2], proto = (byte) (flag & SERIALIZATION_MASK);
// get request id.
long id = Bytes.bytes2long(header, 4);
if ((flag & FLAG_REQUEST) == 0) {
// 如果是Response
...
} else {
// 如果是Request
Request req = new Request(id);
req.setVersion(Version.getProtocolVersion());
req.setTwoWay((flag & FLAG_TWOWAY) != 0);
if ((flag & FLAG_EVENT) != 0) {
req.setEvent(true);
}
try {
Object data;
if (req.isEvent()) {
//如果是Event
...
} else {
//处理Consumer的Request
DecodeableRpcInvocation inv;
if (channel.getUrl().getParameter(DECODE_IN_IO_THREAD_KEY, DEFAULT_DECODE_IN_IO_THREAD)) {
inv = new DecodeableRpcInvocation(channel, req, is, proto);
inv.decode();
} else {
inv = new DecodeableRpcInvocation(channel, req,
new UnsafeByteArrayInputStream(readMessageData(is)), proto);
}
data = inv;
}
req.setData(data);
} catch (Throwable t) {
//错误处理
}
return req;
}
}
服务提供方的DubboCodec在收到Request后，生成DecodeableRpcInvocation对象，然后调用它的decode()方法。最终会进入decodeInvocationArgument()方法：
public static Object decodeInvocationArgument(Channel channel, RpcInvocation inv, Class<?>[] pts, int paraIndex, Object inObject) throws IOException {
// 获取服务端exporter的url
URL url = null;
try {
url = DubboProtocol.getDubboProtocol().getInvoker(channel, inv).getUrl();
} catch (RemotingException e) {
...
return inObject;
}
//判断是否是回调方法
byte callbackstatus = isCallBack(url, inv.getMethodName(), paraIndex);
switch (callbackstatus) {
//如果是回调，生成回调参数的代理
case CallbackServiceCodec.CALLBACK_CREATE:
try {
return referOrDestroyCallbackService(channel, url, pts[paraIndex], inv, Integer.parseInt(inv.getAttachment(INV_ATT_CALLBACK_KEY + paraIndex)), true);
} catch (Exception e) {
logger.error(e.getMessage(), e);
throw new IOException(StringUtils.toString(e));
}
case CallbackServiceCodec.CALLBACK_DESTROY:
try {
return referOrDestroyCallbackService(channel, url, pts[paraIndex], inv, Integer.parseInt(inv.getAttachment(INV_ATT_CALLBACK_KEY + paraIndex)), false);
} catch (Exception e) {
throw new IOException(StringUtils.toString(e));
}
default:
//不是回调，直接返回原参数
return inObject;
}
}
上面的代码中，服务提供端在decode参数时，发现时回调参数，则会返回一个远程调用的代理，生成的过程和Consumer端生成的逻辑时一样的。
private static Object referOrDestroyCallbackService(Channel channel, URL url, Class<?> clazz, Invocation inv, int instid, boolean isRefer) {
Object proxy;
String invokerCacheKey = getServerSideCallbackInvokerCacheKey(channel, clazz.getName(), instid);
String proxyCacheKey = getServerSideCallbackServiceCacheKey(channel, clazz.getName(), instid);
proxy = channel.getAttribute(proxyCacheKey);
String countkey = getServerSideCountKey(channel, clazz.getName());
if (isRefer) {
if (proxy == null) {
//生成回调的url
URL referurl = URL.valueOf("callback://" + url.getAddress() + "/" + clazz.getName() + "?" + INTERFACE_KEY + "=" + clazz.getName());
referurl = referurl.addParametersIfAbsent(url.getParameters()).removeParameter(METHODS_KEY);
if (!isInstancesOverLimit(channel, referurl, clazz.getName(), instid, true)) {
//生成回调的Invoker
@SuppressWarnings("rawtypes")
Invoker<?> invoker = new ChannelWrappedInvoker(clazz, channel, referurl, String.valueOf(instid));
//生成回调的Proxy
proxy = PROXY_FACTORY.getProxy(new AsyncToSyncInvoker<>(invoker));
channel.setAttribute(proxyCacheKey, proxy);
channel.setAttribute(invokerCacheKey, invoker);
increaseInstanceCount(channel, countkey);
//convert error fail fast .
//ignore concurrent problem.
Set<Invoker<?>> callbackInvokers = (Set<Invoker<?>>) channel.getAttribute(CHANNEL_CALLBACK_KEY);
if (callbackInvokers == null) {
callbackInvokers = new ConcurrentHashSet<>(1);
callbackInvokers.add(invoker);
channel.setAttribute(CHANNEL_CALLBACK_KEY, callbackInvokers);
}
}
}
} else {
//Consumer端回调结束，销毁暴露的回调接口
...
}
return proxy;
}
总结
Dubbo对Callback接口跟普通接口的调用对用户是透明的，通过在传输层拦截并复用普通接口的调用逻辑来达到支持callback的目的。
