Dubbo RPC在consumer端是如何跑起来的

DemoService demoService = (DemoService) context.getBean("demoService"); // get remote service proxy
while (true) {
    try {
        String hello = demoService.sayHello("world"); // call remote method
        System.out.println(hello); // get result
        System.in.read();
    } catch (Throwable throwable) {
        throwable.printStackTrace();
    }
}

public interface Directory<T> extends Node {
    Class<T> getInterface();
    List<Invoker<T>> list(Invocation invocation) throws RpcException;
}

@SPI(RandomLoadBalance.NAME)
public interface LoadBalance {
    /**
     * select one invoker in list.
     */
    @Adaptive("loadbalance")
    <T> Invoker<T> select(List<Invoker<T>> invokers, URL url, Invocation invocation) throws RpcException;
}

// DubboInvoker
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) {
        // consumer和provider默认保持一个长连接
        currentClient = clients[0];
    } else { // 如果有多个长连接，则使用轮训方式选择某个连接
        currentClient = clients[index.getAndIncrement() % clients.length];
    }
    try {
        // 默认isAsync为false，isOneWay为false
        boolean isAsync = RpcUtils.isAsync(getUrl(), invocation);
        boolean isAsyncFuture = RpcUtils.isReturnTypeFuture(inv);
        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) {
            // 异步模式，真正的RPC结果处理在ResponseCallback中来做，
            // ResponseCallback在FutureAdapter中设置
            ResponseFuture future = currentClient.request(inv, timeout);
            // For compatibility
            FutureAdapter<Object> futureAdapter = new FutureAdapter<>(future);
            RpcContext.getContext().setFuture(futureAdapter);

            Result result;
            if (isAsyncFuture) {
                // register resultCallback, sometimes we need the async result being processed by the filter chain.
                result = new AsyncRpcResult(futureAdapter, futureAdapter.getResultFuture(), false);
            } else {
                result = new SimpleAsyncRpcResult(futureAdapter, futureAdapter.getResultFuture(), false);
            }
            return result;
        } else {
            // 同步模式，带超时时间，默认1s
            RpcContext.getContext().setFuture(null);
            return (Result) currentClient.request(inv, timeout).get();
        }
    } catch (TimeoutException e) {
    } catch (RemotingException e) {
    }
}

bootstrap.handler(new ChannelInitializer() {
    @Override
    protected void initChannel(Channel ch) throws Exception {
        NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyClient.this);
        ch.pipeline()//.addLast("logging",new LoggingHandler(LogLevel.INFO))//for debug
                .addLast("decoder", adapter.getDecoder())
                .addLast("encoder", adapter.getEncoder()) // encoder编码
                    .addLast("handler", nettyClientHandler);
    }
});

// NettyClientHandler msg是Response类型
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    NettyChannel channel = NettyChannel.getOrAddChannel(ctx.channel(), url, handler);
    try {
        handler.received(channel, msg);
    } finally {
        NettyChannel.removeChannelIfDisconnected(ctx.channel());
    }
}

// .. -> AllChannelHandler
public void received(Channel channel, Object message) throws RemotingException {
    // 响应结果反序列化后，就会提交任务给DubboClientHandler线程池（cexecutor.execute(new ChannelEventRunnable(channel, handler, ChannelState.DISCONNECTED))），
    // 进行后续的结果处理操作
    ExecutorService cexecutor = getExecutorService();
    cexecutor.execute(new ChannelEventRunnable(channel, handler, ChannelState.RECEIVED, message));
}

// HeaderExchangeHandler
public static void received(Channel channel, Response response) {
    try {
        DefaultFuture future = FUTURES.remove(response.getId());
        if (future != null) {
            future.doReceived(response);
        }
    } finally {
        CHANNELS.remove(response.getId());
    }
}

DubboInvoker.doInvoke return (Result) currentClient.request(inv, timeout).get();