dubbo源码分析-consumer端6-数据发送与接收
2016-07-07 00:28
501 查看
consumer端的数据经过处理后,最终进入发送的流程。接下来我们继续跟着数据的流向进行分析。 首先进入到了DubboInvoker,DubboInvoker中包含了多个ExchangeClient, 每个ExchangeClient都对应了一个物理连接,同一个DubboInvoker中的所有ExchangeClient都是连接的同一个ip/port。DubboInvoker循环的从ExchangeClient数组中获取一个,并利用该ExchangeClient发送数据,发送的模式有三种:
1、单项发送:发送完数据直接返回,不需要结果;
2、双向发送:发送完数据后等待数据返回(类似Future.get());
3、异步发送:发送完数据直接返回,同时往RpcContext中存入对应的Future,应用可以通过RpcContext.getContext().getFuture()获取到Future。通过Future可以发起多个异步调用,减少业务的执行时间。
除了异步转同步的功能外,HeaderExchangeClient还加入了心跳检测的功能:
HeaderExchangeChannel:将发送的数据封装为Request对象, 产生一个Future对象(用户异步转同步)与Request关联,然后调用更底层的Channel发送Request。 需要注意的是每一个Request对象都对应了一个唯一id( id为int类型,因此当id达到最大后,又会变为最小值,这样重复利用id)。该id代表了当前连接,在provider有返回数据的时候,会根据这个id来查找对应的Channel。
底层的Channel根据配置不同而不同,默认情况下使用的是netty,consumer端对应实现为NettyClient。netty本身的实现比较高效也很复杂,这里不详讲,有兴趣的同学可以关注本博客内netty相关的文章。这里只关注序列化的部分,具体实现在ExchangeCodec中,以request的encode为例:
循环利用时之前id对应的请求早就已经不再了,因此还是唯一的。
数据返回后的decode方法也是在ExchangeCodec中,具体代码这里不贴了。 response返回时id与request一致,这样可以从consumer缓存map中根据id取出Future并往里设置数据,数据设置完成后,之前在future.get()阻塞的地方恢复(见DubboInvoker的doInvoke方法),继续执行后续逻辑。最终层层返回到业务代码中。
1、单项发送:发送完数据直接返回,不需要结果;
2、双向发送:发送完数据后等待数据返回(类似Future.get());
3、异步发送:发送完数据直接返回,同时往RpcContext中存入对应的Future,应用可以通过RpcContext.getContext().getFuture()获取到Future。通过Future可以发起多个异步调用,减少业务的执行时间。
protected Result doInvoke(final Invocation invocation) throws Throwable { RpcInvocation inv = (RpcInvocation) invocation; final String methodName = RpcUtils.getMethodName(invocation); inv.setAttachment(Constants.PATH_KEY, getUrl().getPath()); inv.setAttachment(Constants.VERSION_KEY, version); // 如果有多个连接则轮流发 ExchangeClient currentClient; if (clients.length == 1) { currentClient = clients[0]; } else { currentClient = clients[index.getAndIncrement() % clients.length]; } try { boolean isAsync = RpcUtils.isAsync(getUrl(), invocation); boolean isOneway = RpcUtils.isOneway(getUrl(), invocation); int timeout = getUrl().getMethodParameter(methodName, Constants.TIMEOUT_KEY,Constants.DEFAULT_TIMEOUT); if (isOneway) { // 不需要返回则发送后不等待立即返回 boolean isSent = getUrl().getMethodParameter(methodName, Constants.SENT_KEY, false); currentClient.send(inv, isSent); RpcContext.getContext().setFuture(null); return new RpcResult(); } else if (isAsync) { // 异步返回时将Future设置到RpcContext中供业务去获取,由业务自行处理异步后的逻辑 ResponseFuture future = currentClient.request(inv, timeout) ; RpcContext.getContext().setFuture(new FutureAdapter<Object>(future)); return new RpcResult(); } else { // 通过future.get()阻塞等待结果返回 RpcContext.getContext().setFuture(null); return (Result) currentClient.request(inv, timeout).get(); } } catch (TimeoutException e) { throw new RpcException(RpcException.TIMEOUT_EXCEPTION, "Invoke remote method timeout. method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e); } catch (RemotingException e) { throw new RpcException(RpcException.NETWORK_EXCEPTION, "Failed to invoke remote method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e); } }
除了异步转同步的功能外,HeaderExchangeClient还加入了心跳检测的功能:
private void startHeatbeatTimer() { // 停止之前的心跳任务 stopHeartbeatTimer(); if ( heartbeat > 0 ) { // 创建定时任务,默认心跳间隔为60s heatbeatTimer = scheduled.scheduleWithFixedDelay( new HeartBeatTask( new HeartBeatTask.ChannelProvider() { public Collection<Channel> getChannels() { return Collections.<Channel>singletonList( HeaderExchangeClient.this ); } }, heartbeat, heartbeatTimeout), heartbeat, heartbeat, TimeUnit.MILLISECONDS ); } }定时任务的逻辑比较简单: 获取连接最后一次读/写数据的时间,如果读或写的时间距当前时间超过心跳的时间,则主动发起一个心跳包,如果该心跳被收到并回复则对应的最后一次读数据时间也会更新,表示连接正常;
HeaderExchangeChannel:将发送的数据封装为Request对象, 产生一个Future对象(用户异步转同步)与Request关联,然后调用更底层的Channel发送Request。 需要注意的是每一个Request对象都对应了一个唯一id( id为int类型,因此当id达到最大后,又会变为最小值,这样重复利用id)。该id代表了当前连接,在provider有返回数据的时候,会根据这个id来查找对应的Channel。
底层的Channel根据配置不同而不同,默认情况下使用的是netty,consumer端对应实现为NettyClient。netty本身的实现比较高效也很复杂,这里不详讲,有兴趣的同学可以关注本博客内netty相关的文章。这里只关注序列化的部分,具体实现在ExchangeCodec中,以request的encode为例:
protected void encodeRequest(Channel channel, ChannelBuffer buffer, Request req) throws IOException { // 加载序列化实现 Serialization serialization = getSerialization(channel); // header. byte[] header = new byte[HEADER_LENGTH]; // set magic number. Bytes.short2bytes(MAGIC, header); // set request and serialization flag. header[2] = (byte) (FLAG_REQUEST | serialization.getContentTypeId()); if (req.isTwoWay()) header[2] |= FLAG_TWOWAY; if (req.isEvent()) header[2] |= FLAG_EVENT; // set request id. Bytes.long2bytes(req.getId(), header, 4); // encode request data. int savedWriteIndex = buffer.writerIndex(); buffer.writerIndex(savedWriteIndex + HEADER_LENGTH); ChannelBufferOutputStream bos = new ChannelBufferOutputStream(buffer); ObjectOutput out = serialization.serialize(channel.getUrl(), bos); if (req.isEvent()) { encodeEventData(channel, out, req.getData()); } else { encodeRequestData(channel, out, req.getData()); } out.flushBuffer(); bos.flush(); bos.close(); int len = bos.writtenBytes(); checkPayload(channel, len); Bytes.int2bytes(len, header, 12); // write buffer.writerIndex(savedWriteIndex); buffer.writeBytes(header); // write header. buffer.writerIndex(savedWriteIndex + HEADER_LENGTH + len); }可以看到一个数据包含header和body, header固定16个字节,前2字节为magic number, 第3字节包括类型(类型+序列化实现的id),第4字节在response时为status,request时为0, 第5-12字节为请求id,13-16字节为body字节数。注意每种序列化都有对应的id,如果要新增序列化方式一定不能与现有id重复。上面的代码中包含一个小细节,一开始并不知道body的字节数,只有序列化完成后才知道,因此header是最后写入的(writerIndex(savedWriteIndex)方法相当于将index移到起点,写入header后再将index移到最后)。每一个请求分配唯一id,到最大后循环使用,
循环利用时之前id对应的请求早就已经不再了,因此还是唯一的。
数据返回后的decode方法也是在ExchangeCodec中,具体代码这里不贴了。 response返回时id与request一致,这样可以从consumer缓存map中根据id取出Future并往里设置数据,数据设置完成后,之前在future.get()阻塞的地方恢复(见DubboInvoker的doInvoke方法),继续执行后续逻辑。最终层层返回到业务代码中。
相关文章推荐
- 从源码安装Mysql/Percona 5.5
- 浅析Ruby的源代码布局及其编程风格
- PHP strtotime函数用法、实现原理和源码分析
- asp.net 抓取网页源码三种实现方法
- jQuery 源码分析笔记(3) Deferred机制
- JS小游戏之仙剑翻牌源码详解
- JS小游戏之宇宙战机源码详解
- jQuery 源码分析笔记(5) jQuery.support
- jQuery-1.9.1源码分析系列(十)事件系统之事件包装
- jQuery源码分析之jQuery中的循环技巧详解
- 本人自用的global.js库源码分享
- java中原码、反码与补码的问题分析
- ASP.NET使用HttpWebRequest读取远程网页源代码
- PHP网页游戏学习之Xnova(ogame)源码解读(六)
- C#获取网页HTML源码实例
- PHP网页游戏学习之Xnova(ogame)源码解读(八)
- PHP网页游戏学习之Xnova(ogame)源码解读(四)
- 深入理解PHP之源码目录结构与功能说明
- JS小游戏之极速快跑源码详解
- JS小游戏之象棋暗棋源码详解