Dubbo中Netty使用形式源码解读

秒速五厘米 2022-05-06 10:44 266阅读 0赞

先从Provider的角度讲，后从Consumer的角度讲，最后合起来讲共同和不同点。

每一个Dubbo接口作为Provider时，都会对外暴露Invoker，执行Export方法：

  1. public class DubboProtocol extends AbstractProtocol { 
  3.      public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException { 
  4.         URL url = invoker.getUrl();
  6.         // 创建 DubboExporter 对象，并添加到 `exporterMap` 。
  7.         // export service.
  8.         String key = serviceKey(url);
  9.         DubboExporter<T> exporter = new DubboExporter<T>(invoker, key, exporterMap);
  10.         exporterMap.put(key, exporter);
  11.         ......
  12.         // 启动服务器
  13.         openServer(url);
  14.         // 初始化序列化优化器
  15.         optimizeSerialization(url);
  16.         return exporter;
  17.     }
  19.      /** * 启动通信服务器 * * @param url URL */
  20.     private void openServer(URL url) { 
  21.         // find server， ip:port
  22.         String key = url.getAddress();
  23.         //client 也可以暴露一个只有server可以调用的服务。
  24.         //client can export a service which's only for server to invoke
  25.         boolean isServer = url.getParameter(Constants.IS_SERVER_KEY, true); // isserver
  26.         if (isServer) { 
  27.             ExchangeServer server = serverMap.get(key);
  28.             if (server == null) { 
  29.                 serverMap.put(key, createServer(url));
  30.             } else { 
  31.                 // server supports reset, use together with override
  32.                 server.reset(url);
  33.             }
  34.         }
  35.     }
  37. }

在此期间需要实例化Server，多个不同的Provider，如果端口相同，那么会共用一个Server(NettyServer)，也就是只实例化一次。

  1. public class NettyServer extends AbstractServer implements Server { 
  3.     @Override
  4.     protected void doOpen() { 
  5.         // 设置日志工厂
  6.         NettyHelper.setNettyLoggerFactory();
  8.         // 实例化 ServerBootstrap
  9.         bootstrap = new ServerBootstrap();
  11.         // 创建线程组
  12.         bossGroup = new NioEventLoopGroup(1, new DefaultThreadFactory("NettyServerBoss", true));
  13.         workerGroup = new NioEventLoopGroup(getUrl().getPositiveParameter(Constants.IO_THREADS_KEY, Constants.DEFAULT_IO_THREADS),
  14.                 new DefaultThreadFactory("NettyServerWorker", true));
  16.         // 创建 NettyServerHandler 对象
  17.         final NettyServerHandler nettyServerHandler = new NettyServerHandler(getUrl(), this);
  18.         // 设置 `channels` 属性
  19.         channels = nettyServerHandler.getChannels();
  21.         bootstrap
  22.                 // 设置它的线程组
  23.                 .group(bossGroup, workerGroup)
  24.                 // 设置 Channel类型
  25.                 .channel(NioServerSocketChannel.class) // Server
  26.                 // 设置可选项
  27.                 .childOption(ChannelOption.TCP_NODELAY, Boolean.TRUE)
  28.                 .childOption(ChannelOption.SO_REUSEADDR, Boolean.TRUE)
  29.                 .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
  30.                 // 设置Handler链路
  31.                 .childHandler(new ChannelInitializer<NioSocketChannel>() { 
  32.                     @Override
  33.                     protected void initChannel(NioSocketChannel ch) { 
  34.                         // 创建 NettyCodecAdapter 对象
  35.                         NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyServer.this);
  36.                         ch.pipeline()//.addLast("logging",new LoggingHandler(LogLevel.INFO))//for debug
  37.                                 .addLast("decoder", adapter.getDecoder()) // 解码
  38.                                 .addLast("encoder", adapter.getEncoder())  // 编码
  39.                                 .addLast("handler", nettyServerHandler); // 处理器
  40.                     }
  41.                 });
  43.         // 服务器绑定端口监听
  44.         // bind
  45.         ChannelFuture channelFuture = bootstrap.bind(getBindAddress());
  46.         channelFuture.syncUninterruptibly();
  47.         channel = channelFuture.channel();
  48.     }
  50. }

Netty的Reactor模型和ChannelPipeline模型网上很多，这里就不列举了。

上述代码实例化Netty的Reactor模型相关的对象，默认使用NioSocketChannel（Selector）作为数据传输通道，按顺序依次添加日志，解码，编码，业务处理的Handler。

Provider的Server在接收到Consumer（Client）的请求数据后，先通过解码Handler进行以下处理：

  1. 首先通过Dubbo协议的数据格式来对TCP/IP数据进行拆包
  2. 将拆包得到的数据进行反序列化，比如Hession，Json，Thrift，ProtoBuf等，得到Java DTO
  3. 将DTO以及Invoker的相关信息封装为一个Task，投递到执行线程池中，最终调用Dubbo接口
  4. 将业务结果通过序列化工具序列化为可传输的字节数组，然后根据传输协议(Dubbo)组装成相应的数据格式，写入Channel

1.数据拆包及获取传输的Body：

Dubbo协议数据格式

  1. public class ExchangeCodec extends TelnetCodec { 
  3.     // header length.
  4.     protected static final int HEADER_LENGTH = 16;
  5.     // magic header.
  6.     protected static final short MAGIC = (short) 0xdabb;
  7.     protected static final byte MAGIC_HIGH = Bytes.short2bytes(MAGIC)[0];
  8.     protected static final byte MAGIC_LOW = Bytes.short2bytes(MAGIC)[1];
  10.      public Object decode(Channel channel, ChannelBuffer buffer) throws IOException { 
  11.         // 读取 Header 数组，Header长度为16字节
  12.         int readable = buffer.readableBytes();
  13.         byte[] header = new byte[Math.min(readable, HEADER_LENGTH)];
  14.         buffer.readBytes(header);
  15.         // 解码
  16.         return decode(channel, buffer, readable, header);
  17.     }
  19.        /** * @param channel * @param buffer * @param readable 可读取长度 * @param header 数据头部 * @return * @throws IOException */
  20.     @Override
  21.     protected Object decode(Channel channel, ChannelBuffer buffer, int readable, byte[] header) throws IOException { 
  22.         // 非 Dubbo 协议，目前是 Telnet 命令。
  23.         // 通过MAGIC判断请求数据是否为Dubbo协议
  24.         if (readable > 0 && header[0] != MAGIC_HIGH || readable > 1 && header[1] != MAGIC_LOW) { 
  25.             // 将 buffer 完全复制到 `header` 数组中。因为，上面的 `#decode(channel, buffer)` 方法，可能未读全
  26.             int length = header.length;
  27.             if (header.length < readable) { 
  28.                 header = Bytes.copyOf(header, readable);
  29.                 buffer.readBytes(header, length, readable - length);
  30.             }
  31.             // 【TODO 8026 】header[i] == MAGIC_HIGH && header[i + 1] == MAGIC_LOW ？
  32.             for (int i = 1; i < header.length - 1; i++) { 
  33.                 if (header[i] == MAGIC_HIGH && header[i + 1] == MAGIC_LOW) { 
  34.                     buffer.readerIndex(buffer.readerIndex() - header.length + i);
  35.                     header = Bytes.copyOf(header, i);
  36.                     break;
  37.                 }
  38.             }
  39.             // 提交给父类( Telnet ) 处理，目前是 Telnet 命令。
  40.             return super.decode(channel, buffer, readable, header);
  41.         }
  42.         // Header 长度不够，返回需要更多的输入
  43.         // check length.
  44.         if (readable < HEADER_LENGTH) { 
  45.             return DecodeResult.NEED_MORE_INPUT;
  46.         }
  48.         // `[96 - 127]`：Body 的**长度**。通过该长度，读取 Body 。
  49.         // get data length.
  50.         int len = Bytes.bytes2int(header, 12);
  51.         checkPayload(channel, len);
  53.         // 总长度不够，返回需要更多的输入
  54.         int tt = len + HEADER_LENGTH;
  55.         if (readable < tt) { 
  56.             return DecodeResult.NEED_MORE_INPUT;
  57.         }
  59.         // 解析 Header + Body
  60.         // limit input stream.
  61.         ChannelBufferInputStream is = new ChannelBufferInputStream(buffer, len);
  62.         try { 
  63.             return decodeBody(channel, is, header);
  64.         } finally { 
  65.             // skip 未读完的流，并打印错误日志
  66.             if (is.available() > 0) { 
  67.                 try { 
  68.                     if (logger.isWarnEnabled()) { 
  69.                         logger.warn("Skip input stream " + is.available());
  70.                     }
  71.                     StreamUtils.skipUnusedStream(is);
  72.                 } catch (IOException e) { 
  73.                     logger.warn(e.getMessage(), e);
  74.                 }
  75.             }
  76.         }
  77.     }
  79. }

每次请求的数据都严格按照指定协议组装，Header+Body。Header16个字节长度，通过Header的数据结构来判断Channel里读取的数据是不是一次请求的完整数据。如果不够，那么返回，等待此Channel的下次数据读取；如果够了，那么将ByteBufer里指定Body长度的数据提交给序列化器反序列化为Java DTO。

2.将Body反序列化为Java DTO：

  1. public class ExchangeCodec extends TelnetCodec { 
  3.     /** * 解析，返回 Request 或 Response * * @param channel 通道 * @param is 输出 * @param header Header * @return 结果 * @throws IOException 当发生 IO 异常时 */
  4.     protected Object decodeBody(Channel channel, InputStream is, byte[] header) throws IOException { 
  5.         byte flag = header[2], proto = (byte) (flag & SERIALIZATION_MASK);
  6.         Serialization s = CodecSupport.getSerialization(channel.getUrl(), proto);
  7.         ObjectInput in = s.deserialize(channel.getUrl(), is);
  8.         // Response
  9.         // get request id.
  10.         long id = Bytes.bytes2long(header, 4);
  11.         if ((flag & FLAG_REQUEST) == 0) {  // Response
  12.             // decode response.
  13.             Response res = new Response(id);
  14.             if ((flag & FLAG_EVENT) != 0) { 
  15.                 res.setEvent(Response.HEARTBEAT_EVENT);
  16.             }
  17.             // get status.
  18.             byte status = header[3];
  19.             res.setStatus(status);
  20.             if (status == Response.OK) { 
  21.                 try { 
  22.                     Object data;
  23.                     if (res.isHeartbeat()) { 
  24.                         data = decodeHeartbeatData(channel, in);
  25.                     } else if (res.isEvent()) { 
  26.                         data = decodeEventData(channel, in);
  27.                     } else { 
  28.                         data = decodeResponseData(channel, in, getRequestData(id)); // `#getRequestData(id)` 的调用，是多余的
  29.                     }
  30.                     res.setResult(data);
  31.                 } catch (Throwable t) { 
  32.                     res.setStatus(Response.CLIENT_ERROR);
  33.                     res.setErrorMessage(StringUtils.toString(t));
  34.                 }
  35.             } else { 
  36.                 res.setErrorMessage(in.readUTF());
  37.             }
  38.             return res;
  39.         // Request
  40.         } else {  // Request
  41.             // decode request.
  42.             Request req = new Request(id);
  43.             req.setVersion("2.0.0");
  44.             req.setTwoWay((flag & FLAG_TWOWAY) != 0);
  45.             if ((flag & FLAG_EVENT) != 0) {  // 心跳事件
  46.                 req.setEvent(Request.HEARTBEAT_EVENT);
  47.             }
  48.             try { 
  49.                 Object data;
  50.                 if (req.isHeartbeat()) { 
  51.                     data = decodeHeartbeatData(channel, in);
  52.                 } else if (req.isEvent()) { 
  53.                     data = decodeEventData(channel, in);
  54.                 } else { 
  55.                     data = decodeRequestData(channel, in);
  56.                 }
  57.                 req.setData(data);
  58.             } catch (Throwable t) { 
  59.                 // bad request
  60.                 req.setBroken(true);
  61.                 req.setData(t);
  62.             }
  63.             return req;
  64.         }
  65.     }
  67. }

根据Header里的第三个字节来判断当前数据是请求还是响应（因为Server和Client共用一套数据处理流程），实例化响应的Request，Response对象，将Header的第四个字节Status存入返回对象中，将第4-12个字节组成的RequestId存入返回对象，将Body数据通过反序列化得到的DTO存入返回对象。RequestId是一次请求的唯一标识，Client通过此标识来将Server返回的数据映射到之前发送的请求上。

3.对于Server，异步形式处理请求数据

  1. public class AllChannelHandler extends WrappedChannelHandler { 
  3.      public void received(Channel channel, Object message) throws RemotingException { 
  4.         ExecutorService cexecutor = getExecutorService();
  5.         try { 
  6.             cexecutor.execute(new ChannelEventRunnable(channel, handler, ChannelState.RECEIVED, message));
  7.         } catch (Throwable t) { 
  8.            ......
  9.         }
  10.     }
  12. }

4.调用Dubbo接口，将请求结果反序列化，然后按照协议格式组装数据写入Channel:

调用Dubbo接口的代码在 DubboProtocol#requestHandler里，代码很容易看明白，这里就不列举了
编码的Handler执行顺序在业务Handler写入数据之后，其会请求结果序列化，然后按照协议格式数据

  1. public class ExchangeCodec extends TelnetCodec { 
  3.      public void encode(Channel channel, ChannelBuffer buffer, Object msg) throws IOException { 
  4.         if (msg instanceof Request) {  // 请求
  5.             encodeRequest(channel, buffer, (Request) msg);
  6.         } else if (msg instanceof Response) {  // 响应
  7.             encodeResponse(channel, buffer, (Response) msg);
  8.         } else {  // 提交给父类( Telnet ) 处理，目前是 Telnet 命令的结果。
  9.             super.encode(channel, buffer, msg);
  10.         }
  11.     }
  13.     /** * 编码响应 * * @param channel 通道 * @param buffer Buffer * @param res 响应 * @throws IOException 当发生 IO 异常时 */
  14.     protected void encodeResponse(Channel channel, ChannelBuffer buffer, Response res) throws IOException { 
  15.         int savedWriteIndex = buffer.writerIndex();
  16.         try { 
  17.             Serialization serialization = getSerialization(channel);
  18.             // `[0, 15]`：Magic Number
  19.             // 构建Header
  20.             byte[] header = new byte[HEADER_LENGTH];
  21.             // set magic number.
  22.             Bytes.short2bytes(MAGIC, header);
  24.             // `[16, 20]`：Serialization 编号 && `[23]`：响应。
  25.             // set request and serialization flag.
  26.             header[2] = serialization.getContentTypeId();
  28.             // `[21]`：`event` 是否为事件。
  29.             if (res.isHeartbeat()) header[2] |= FLAG_EVENT;
  31.             // `[24 - 31]`：`status` 状态。
  32.             // set response status.
  33.             byte status = res.getStatus();
  34.             header[3] = status;
  36.             // `[32 - 95]`：`id` 编号，Long 型。
  37.             // set request id.
  38.             Bytes.long2bytes(res.getId(), header, 4);
  40.             // 编码 `Request.data` 到 Body ，并写入到 Buffer
  41.             buffer.writerIndex(savedWriteIndex + HEADER_LENGTH);
  42.             ChannelBufferOutputStream bos = new ChannelBufferOutputStream(buffer); // 序列化 Output
  43.             ObjectOutput out = serialization.serialize(channel.getUrl(), bos);
  44.             // encode response data or error message.
  45.             if (status == Response.OK) { 
  46.                 if (res.isHeartbeat()) { 
  47.                     encodeHeartbeatData(channel, out, res.getResult());
  48.                 } else { 
  49.                     encodeResponseData(channel, out, res.getResult());
  50.                 }
  51.             } else { 
  52.                 out.writeUTF(res.getErrorMessage());
  53.             }
  54.             // 释放资源
  55.             out.flushBuffer();
  56.             if (out instanceof Cleanable) { 
  57.                 ((Cleanable) out).cleanup();
  58.             }
  59.             bos.flush();
  60.             bos.close();
  62.             // 检查 Body 长度，是否超过消息上限。
  63.             int len = bos.writtenBytes();
  64.             checkPayload(channel, len);
  65.             // `[96 - 127]`：Body 的**长度**。
  66.             Bytes.int2bytes(len, header, 12);
  68.             // 写入 Header 到 Buffer
  69.             // write
  70.             buffer.writerIndex(savedWriteIndex);
  71.             buffer.writeBytes(header); // write header.
  72.             buffer.writerIndex(savedWriteIndex + HEADER_LENGTH + len);
  73.         } catch (Throwable t) { 
  74.            ......          
  75.         }
  76.     }
  77. }

在Provider端的Netty的作用流程基本就如上述所示，下面将Consumer端相对于Provider端不同的作用流程。

每个接口的Consumer在初始化时都会实例化一个NettyClient，也就是实例化一个Netty的Bootstrap。如果按照传统思路，每个Bootstrap关联一个EventLoopGroup，内部持有一个Selector，那么每个接口Consumer实例都会有一个Selector，这会造成资源的很大浪费，所以Dubbp定义了一个EventLoopGroup类型的常量，让所有Consumer在实例化时都用同一个EventLoopGroup，也就是一个Selector处理所有的Consumer的Channel：

  1. public class NettyClient extends AbstractClient { 
  3.     private static final NioEventLoopGroup nioEventLoopGroup = new NioEventLoopGroup(Constants.DEFAULT_IO_THREADS, new DefaultThreadFactory("NettyClientWorker", true));
  5.     protected void doOpen() { 
  6.         // 设置日志工厂
  7.         NettyHelper.setNettyLoggerFactory();
  9.         // 创建 NettyClientHandler 对象
  10.         final NettyClientHandler nettyClientHandler = new NettyClientHandler(getUrl(), this);
  12.         // 实例化 ServerBootstrap
  13.         bootstrap = new Bootstrap();
  14.         bootstrap
  15.                 // 设置它的线程组
  16.                 .group(nioEventLoopGroup)
  17.                 // 设置可选项
  18.                 .option(ChannelOption.SO_KEEPALIVE, true)
  19.                 .option(ChannelOption.TCP_NODELAY, true)
  20.                 .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
  21.                 //.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, getTimeout())
  22.                 // 设置 Channel类型
  23.                 .channel(NioSocketChannel.class);
  25.         // 设置连接超时时间
  26.         if (getTimeout() < 3000) { 
  27.             bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000);
  28.         } else { 
  29.             bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, getTimeout());
  30.         }
  32.         // 设置责Handler链路
  33.         bootstrap.handler(new ChannelInitializer() { 
  34.             @Override
  35.             protected void initChannel(Channel ch) { 
  36.                 // 创建 NettyCodecAdapter 对象
  37.                 NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyClient.this);
  38.                 ch.pipeline()//.addLast("logging",new LoggingHandler(LogLevel.INFO))//for debug
  39.                         .addLast("decoder", adapter.getDecoder()) // 解码
  40.                         .addLast("encoder", adapter.getEncoder()) // 编码
  41.                         .addLast("handler", nettyClientHandler); // 处理器
  42.             }
  43.         });
  44.     }
  45. }

Consumer相比于Provider还有一点不同：
Provider会使用线程池异步的处理Consumer发送的请求，处理完成后将相应数据写入Channel；
Consumer在大部分情况下都是发送同步请求，也就是会阻塞当前线程，直到Provider返回数据，或者阻塞超时。

  1. public class DubboInvoker<T> extends AbstractInvoker<T> { 
  3.        protected Result doInvoke(final Invocation invocation) { 
  4.         RpcInvocation inv = (RpcInvocation) invocation;
  5.         // 获得方法名
  6.         final String methodName = RpcUtils.getMethodName(invocation);
  7.         // 获得 `path`( 服务名 )，`version`
  8.         inv.setAttachment(Constants.PATH_KEY, getUrl().getPath());
  9.         inv.setAttachment(Constants.VERSION_KEY, version);
  11.         // 获得 ExchangeClient 对象
  12.         ExchangeClient currentClient;
  13.         if (clients.length == 1) { 
  14.             currentClient = clients[0];
  15.         } else { 
  16.             currentClient = clients[index.getAndIncrement() % clients.length];
  17.         }
  18.         // 远程调用
  19.         try { 
  20.             // 获得是否异步调用
  21.             boolean isAsync = RpcUtils.isAsync(getUrl(), invocation);
  22.             // 获得是否单向调用
  23.             boolean isOneway = RpcUtils.isOneway(getUrl(), invocation);
  24.             // 获得超时时间
  25.             int timeout = getUrl().getMethodParameter(methodName, Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);
  26.             // 单向调用
  27.             if (isOneway) { 
  28.                 boolean isSent = getUrl().getMethodParameter(methodName, Constants.SENT_KEY, false);
  29.                 currentClient.send(inv, isSent);
  30.                 RpcContext.getContext().setFuture(null);
  31.                 return new RpcResult();
  32.             // 异步调用
  33.             } else if (isAsync) { 
  34.                 ResponseFuture future = currentClient.request(inv, timeout);
  35.                 RpcContext.getContext().setFuture(new FutureAdapter<Object>(future));
  36.                 return new RpcResult();
  37.             // 同步调用
  38.             } else { 
  39.                 RpcContext.getContext().setFuture(null);
  40.                 return (Result) currentClient.request(inv, timeout).get();
  41.             }
  42.         } catch (TimeoutException e) { 
  43.             throw new RpcException(RpcException.TIMEOUT_EXCEPTION, "Invoke remote method timeout. method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
  44.         } catch (RemotingException e) { 
  45.             throw new RpcException(RpcException.NETWORK_EXCEPTION, "Failed to invoke remote method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
  46.         }
  47.     }
  49. }

如果是同步调用的情况下，会将调用相关信息存入一个DefaultFuture中，然后阻塞，等待Provider返回信息，依据requestId来唤醒当前线程：

  1. public class DefaultFuture implements ResponseFuture { 
  3.     /** * 通道集合 * * key：请求编号 */
  4.     private static final Map<Long, Channel> CHANNELS = new ConcurrentHashMap<Long, Channel>();
  5.     /** * Future 集合 * * key：请求编号 */
  6.     private static final Map<Long, DefaultFuture> FUTURES = new ConcurrentHashMap<Long, DefaultFuture>();
  8.     public DefaultFuture(Channel channel, Request request, int timeout) { 
  9.         this.channel = channel;
  10.         this.request = request;
  11.         this.id = request.getId();
  12.         this.timeout = timeout > 0 ? timeout : channel.getUrl().getPositiveParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);
  13.         // put into waiting map.
  14.         FUTURES.put(id, this);
  15.         CHANNELS.put(id, channel);
  16.     }
  17.     /** * 阻塞式的等待结果 * * @param timeout * @return * @throws RemotingException */
  18.     public Object get(int timeout) throws RemotingException { 
  19.         if (timeout <= 0) { 
  20.             timeout = Constants.DEFAULT_TIMEOUT;
  21.         }
  22.         // 若未完成，等待
  23.         if (!isDone()) { 
  24.             long start = System.currentTimeMillis();
  25.             lock.lock();
  26.             try { 
  27.                 // 等待完成或超时
  28.                 while (!isDone()) { 
  29.                     done.await(timeout, TimeUnit.MILLISECONDS);
  30.                     if (isDone() || System.currentTimeMillis() - start > timeout) { 
  31.                         break;
  32.                     }
  33.                 }
  34.             } catch (InterruptedException e) { 
  35.                 throw new RuntimeException(e);
  36.             } finally { 
  37.                 lock.unlock();
  38.             }
  39.             // 未完成，抛出超时异常 TimeoutException
  40.             if (!isDone()) { 
  41.                 throw new TimeoutException(sent > 0, channel, getTimeoutMessage(false));
  42.             }
  43.         }
  44.         // 返回响应
  45.         return returnFromResponse();
  46.     }
  48. }

Consumer解析处Provider传递回来的数据，每解析成功一个返回结果，如果此次请求时同步请求，就会通过Condition唤醒相应requestId的请求线程，此时同步请求才会有结果返回。

  1. /** * 完成 Condition */
  2.     private final Condition done = lock.newCondition();
  4.     /** * 接收响应( Response ) * * @param channel 通道 * @param response 响应 */
  5.     public static void received(Channel channel, Response response) { 
  6.         try { 
  7.             // 移除 FUTURES
  8.             DefaultFuture future = FUTURES.remove(response.getId());
  9.             // 接收结果
  10.             if (future != null) { 
  11.                 future.doReceived(response);
  12.             } else { 
  13.                 logger.warn("The timeout response finally returned at "
  14.                     + (new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date()))
  15.                     + ", response " + response
  16.                     + (channel == null ? "" : ", channel: " + channel.getLocalAddress()
  17.                     + " -> " + channel.getRemoteAddress()));
  18.             }
  19.             // 移除 CHANNELS
  20.         } finally { 
  21.             CHANNELS.remove(response.getId());
  22.         }
  23.     }
  25.     private void doReceived(Response res) { 
  26.         // 锁定
  27.         lock.lock();
  28.         try { 
  29.             // 设置结果
  30.             response = res;
  31.             // 通知，唤醒等待
  32.             if (done != null) { 
  33.                 done.signal();
  34.             }
  35.         } finally { 
  36.             // 释放锁定
  37.             lock.unlock();
  38.         }
  39.         // 调用回调
  40.         if (callback != null) { 
  41.             invokeCallback(callback);
  42.         }
  43.     }

Consumer的拆包，序列化，协议格式数据组装都和Provider都是相同的，每次请求都会有一个唯一标识requestId来匹配请求和响应。请求在Provider端使用线程池异步执行，在Consumer端可同步，异步，Oneway形式，如果是同步则阻塞线程，等待指定requestId的响应返回来唤醒线程。

发表评论

表情：
评论列表 （有 0 条评论，266人围观）

还没有评论，来说两句吧...

相关阅读