Consumer Provider

消费者使用 NettyClient，提供者使用 NettyServer



Consumer端 

com.alibaba.dubbo.rpc.protocol.dubbo.DubboProtocol
com.alibaba.dubbo.rpc.protocol.dubbo.DubboProtocol.getClients(URL)

private ExchangeClient[] getClients(URL url){
    //是否共享连接
    boolean service_share_connect = false;
    int connections = url.getParameter(Constants.CONNECTIONS_KEY, 0);
    //如果connections不配置，则共享连接，否则每服务每连接
    if (connections == 0){
        service_share_connect = true;
        connections = 1;
    }
        
    ExchangeClient[] clients = new ExchangeClient[connections];
    for (int i = 0; i < clients.length; i++) {
        if (service_share_connect){
            clients[i] = getSharedClient(url);
        } else {
            clients[i] = initClient(url);
        }
    }
    return clients;
}

com.alibaba.dubbo.remoting.transport.netty.NettyClient

com.alibaba.dubbo.remoting.exchange.support.header.HeaderExchangeChannel 


String result = helloService.hello("dyyx");
com.alibaba.dubbo.remoting.exchange.support.header.HeaderExchangeChannel.request(Object, int)
RpcInvocation [methodName=hello, parameterTypes=[class java.lang.String], arguments=[dyyx], attachments={path=dyyx.HelloService, interface=dyyx.HelloService, version=1.0.0, group=group1}]

com.alibaba.dubbo.remoting.transport.DecodeHandler.decode(Object)

结果异步转同步处理

com.alibaba.dubbo.remoting.exchange.support.DefaultFuture


    public static void received(Channel channel, Response response) {
        try {
            DefaultFuture future = FUTURES.remove(response.getId());
            if (future != null) {
                future.doReceived(response);
            } else {
                logger.warn("The timeout response finally returned at " 
                            + (new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date())) 
                            + ", response " + response 
                            + (channel == null ? "" : ", channel: " + channel.getLocalAddress() 
                                + " -> " + channel.getRemoteAddress()));
            }
        } finally {
            CHANNELS.remove(response.getId());
        }
    }

    private void doReceived(Response res) {
        lock.lock();
        try {
            response = res;
            if (done != null) {
                done.signal();
            }
        } finally {
            lock.unlock();
        }
        if (callback != null) {
            invokeCallback(callback);
        }
    }
    
    


Thread [main] (Suspended (breakpoint at line 102 in HeaderExchangeChannel)) 
    HeaderExchangeChannel.request(Object, int) line: 102    
    HeaderExchangeClient.request(Object, int) line: 91  
    ReferenceCountExchangeClient.request(Object, int) line: 81  
    DubboInvoker<T>.doInvoke(Invocation) line: 96   
    DubboInvoker<T>(AbstractInvoker<T>).invoke(Invocation) line: 144    
    ListenerInvokerWrapper<T>.invoke(Invocation) line: 74   
    MonitorFilter.invoke(Invoker<?>, Invocation) line: 75   
    ProtocolFilterWrapper$1.invoke(Invocation) line: 91 
    FutureFilter.invoke(Invoker<?>, Invocation) line: 53    
    ProtocolFilterWrapper$1.invoke(Invocation) line: 91 
    ConsumerContextFilter.invoke(Invoker<?>, Invocation) line: 48   
    ProtocolFilterWrapper$1.invoke(Invocation) line: 91 
    InvokerInvocationHandler.invoke(Object, Method, Object[]) line: 52  
    proxy0.hello(String) line: not available    
    ClientMain2.main(String[]) line: 24 


结果异步转同步处理

Daemon Thread [DubboClientHandler-127.0.0.1:20880-thread-1] (Suspended (breakpoint at line 256 in DefaultFuture))   
    DefaultFuture.doReceived(Response) line: 256    
    DefaultFuture.received(Channel, Response) line: 240 
    HeaderExchangeHandler.handleResponse(Channel, Response) line: 96    
    HeaderExchangeHandler.received(Channel, Object) line: 177   
    DecodeHandler.received(Channel, Object) line: 52    
    ChannelEventRunnable.run() line: 82 
    ThreadPoolExecutor.runWorker(ThreadPoolExecutor$Worker) line: 1142  
    ThreadPoolExecutor$Worker.run() line: 617   
    Thread.run() line: 745  






Provider 端

com.alibaba.dubbo.remoting.transport.DecodeHandler.received(Channel, Object)

com.alibaba.dubbo.remoting.exchange.support.header.HeaderExchangeHandler.received(Channel, Object)
Request [id=7, version=2.0.0, twoway=true, event=false, broken=false, data=RpcInvocation [methodName=hello, parameterTypes=[class java.lang.String], arguments=[dyyx], attachments={path=dyyx.HelloService, input=168, dubbo=2.8.4, interface=dyyx.HelloService, version=1.0.0, timeout=500, group=group1}]]
data=RpcInvocation [methodName=hello, parameterTypes=[class java.lang.String], arguments=[dyyx], attachments={path=dyyx.HelloService, input=168, dubbo=2.8.4, interface=dyyx.HelloService, version=1.0.0, timeout=500, group=group1}

com.alibaba.dubbo.remoting.exchange.support.header.HeaderExchangeHandler.handleRequest(ExchangeChannel, Request)

发送响应结果
com.alibaba.dubbo.remoting.transport.netty.NettyHandler.writeRequested(ChannelHandlerContext, MessageEvent)

[id: 0xc89e46a2, /192.168.1.102:50409 => /192.168.1.102:20880] WRITE: Response [id=31, version=2.0.0, status=20, event=false, error=null, result=RpcResult [result=hello,dyyx,2021-01-10T11:12:44.071, exception=null]]




Daemon Thread [DubboServerHandler-192.168.1.102:20880-thread-4] (Suspended (breakpoint at line 31 in HelloServiceImpl)) 
    HelloServiceImpl.hello(String) line: 31 
    Wrapper1.invokeMethod(Object, String, Class[], Object[]) line: not available    
    JavassistProxyFactory$1.doInvoke(T, String, Class<?>[], Object[]) line: 46  
    JavassistProxyFactory$1(AbstractProxyInvoker<T>).invoke(Invocation) line: 72    
    RegistryProtocol$InvokerDelegete<T>(InvokerWrapper<T>).invoke(Invocation) line: 53  
    ExceptionFilter.invoke(Invoker<?>, Invocation) line: 64 
    ProtocolFilterWrapper$1.invoke(Invocation) line: 91 
    MonitorFilter.invoke(Invoker<?>, Invocation) line: 65   
    ProtocolFilterWrapper$1.invoke(Invocation) line: 91 
    TimeoutFilter.invoke(Invoker<?>, Invocation) line: 42   
    ProtocolFilterWrapper$1.invoke(Invocation) line: 91 
    TraceFilter.invoke(Invoker<?>, Invocation) line: 78 
    ProtocolFilterWrapper$1.invoke(Invocation) line: 91 
    ContextFilter.invoke(Invoker<?>, Invocation) line: 70   
    ProtocolFilterWrapper$1.invoke(Invocation) line: 91 
    GenericFilter.invoke(Invoker<?>, Invocation) line: 132  
    ProtocolFilterWrapper$1.invoke(Invocation) line: 91 
    ClassLoaderFilter.invoke(Invoker<?>, Invocation) line: 38   
    ProtocolFilterWrapper$1.invoke(Invocation) line: 91 
    EchoFilter.invoke(Invoker<?>, Invocation) line: 38  
    ProtocolFilterWrapper$1.invoke(Invocation) line: 91 
    DubboProtocol$1.reply(ExchangeChannel, Object) line: 113    
    HeaderExchangeHandler.handleRequest(ExchangeChannel, Request) line: 84  
    HeaderExchangeHandler.received(Channel, Object) line: 170   
    DecodeHandler.received(Channel, Object) line: 52    
    ChannelEventRunnable.run() line: 82 
    ThreadPoolExecutor.runWorker(ThreadPoolExecutor$Worker) line: 1142  
    ThreadPoolExecutor$Worker.run() line: 617   
    Thread.run() line: 745  
    
    
Daemon Thread [DubboServerHandler-192.168.1.102:20880-thread-10] (Suspended (breakpoint at line 102 in NettyHandler))   
    NettyHandler.writeRequested(ChannelHandlerContext, MessageEvent) line: 102  
    NettyHandler(SimpleChannelHandler).handleDownstream(ChannelHandlerContext, ChannelEvent) line: 254  
    DefaultChannelPipeline.sendDownstream(DefaultChannelPipeline$DefaultChannelHandlerContext, ChannelEvent) line: 591  
    DefaultChannelPipeline.sendDownstream(ChannelEvent) line: 582   
    Channels.write(Channel, Object, SocketAddress) line: 704    
    Channels.write(Channel, Object) line: 671   
    NioAcceptedSocketChannel(AbstractChannel).write(Object) line: 248   
    NettyChannel.send(Object, boolean) line: 98 
    NettyChannel(AbstractPeer).send(Object) line: 51    
    HeaderExchangeHandler.received(Channel, Object) line: 171   
    DecodeHandler.received(Channel, Object) line: 52    
    ChannelEventRunnable.run() line: 82 
    ThreadPoolExecutor.runWorker(ThreadPoolExecutor$Worker) line: 1142  
    ThreadPoolExecutor$Worker.run() line: 617   
    Thread.run() line: 745