dubbo 请求发送与响应设计

逃离我推掉我的手 2022-04-18 01:26 168阅读 0赞

# dubbo invoke流程 #

com.alibaba.dubbo.rpc.protocol.AbstractInvoker#invoke
    	>com.alibaba.dubbo.rpc.protocol.dubbo.DubboInvoker#doInvoke
    		>区分isOneway/isAsync/同步调用
    		>同步调用 currentClient.request(inv, timeout).get()
    		>com.alibaba.dubbo.remoting.exchange.ExchangeChannel#request(java.lang.Object, int)
    		>com.alibaba.dubbo.remoting.exchange.ResponseFuture#get()
    			>com.alibaba.dubbo.remoting.exchange.support.header.HeaderExchangeClient#request(java.lang.Object, int)
    				>com.alibaba.dubbo.remoting.exchange.support.header.HeaderExchangeChannel#request(java.lang.Object, int)
    					>构造请求future用来跟踪响应 DefaultFuture future = new DefaultFuture(channel, req, timeout);
    					>com.alibaba.dubbo.remoting.exchange.support.header.HeaderExchangeChannel#send(java.lang.Object)
    						>com.alibaba.dubbo.remoting.transport.netty.NettyChannel#send
    							>netty 发送请求 ChannelFuture future = channel.write(message);
    							>success = future.await(timeout);

# DefaultFuture设计 #

## 成员变量 ##

// 记录请求ID对应Channel（即服务端信息）
    	private static final Map<Long, Channel>       CHANNELS   = new ConcurrentHashMap<Long, Channel>();
    	// 记录请求ID对应Future
        private static final Map<Long, DefaultFuture> FUTURES   = new ConcurrentHashMap<Long, DefaultFuture>();
    
        // invoke id. 请求标志ID
        private final long                            id;
    
        private final Channel                         channel;
        
        private final Request                         request;
    
        private final int                             timeout;
    
        private final Lock                            lock = new ReentrantLock();
    	
    	// 完成条件
        private final Condition                       done = lock.newCondition();
    
        private final long                            start = System.currentTimeMillis();
    
        private volatile long                         sent;
        // 为何设计成volatile？是否有可能服务端发了多次响应
        private volatile Response                     response;
    
        private volatile ResponseCallback             callback;

## 构造函数 ##

public DefaultFuture(Channel channel, Request request, int timeout){ 
            this.channel = channel;
            this.request = request;
            this.id = request.getId();
            this.timeout = timeout > 0 ? timeout : channel.getUrl().getPositiveParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);
            // put into waiting map. 加入等待Map
            FUTURES.put(id, this);
            CHANNELS.put(id, channel);
        }

## 同步调用时ResponseFuture\#get() ##

public Object get() throws RemotingException { 
            return get(timeout);
        }
    
        public Object get(int timeout) throws RemotingException { 
            if (timeout <= 0) { 
                timeout = Constants.DEFAULT_TIMEOUT;
            }
            if (! isDone()) { 
                long start = System.currentTimeMillis();
                lock.lock();
                try { 
                    while (! isDone()) { 
                    	// 线程休眠，让出CPU
                        done.await(timeout, TimeUnit.MILLISECONDS);
                        if (isDone() || System.currentTimeMillis() - start > timeout) { 
                            break;
                        }
                    }
                } catch (InterruptedException e) { 
                    throw new RuntimeException(e);
                } finally { 
                    lock.unlock();
                }
                if (! isDone()) { 
                    throw new TimeoutException(sent > 0, channel, getTimeoutMessage(false));
                }
            }
            return returnFromResponse();
        }

## 检查等待map是否超时 ##

private static class RemotingInvocationTimeoutScan implements Runnable { 
    
            public void run() { 
                while (true) { 
                    try { 
                        for (DefaultFuture future : FUTURES.values()) { 
                            if (future == null || future.isDone()) { 
                                continue;
                            }
                            if (System.currentTimeMillis() - future.getStartTimestamp() > future.getTimeout()) { 
                                // create exception response. 如果超时，则构造超时响应Response
                                Response timeoutResponse = new Response(future.getId());
                                // set timeout status.
                                timeoutResponse.setStatus(future.isSent() ? Response.SERVER_TIMEOUT : Response.CLIENT_TIMEOUT);
                                timeoutResponse.setErrorMessage(future.getTimeoutMessage(true));
                                // handle response.
                                DefaultFuture.received(future.getChannel(), timeoutResponse);
                            }
                        }
                        Thread.sleep(30);
                    } catch (Throwable e) { 
                        logger.error("Exception when scan the timeout invocation of remoting.", e);
                    }
                }
            }
        }

## 接收到服务端响应处理 ##

com.alibaba.dubbo.remoting.exchange.support.header.HeaderExchangeHandler#received
    	>com.alibaba.dubbo.remoting.exchange.support.header.HeaderExchangeHandler#handleResponse
    		>com.alibaba.dubbo.remoting.exchange.support.DefaultFuture#received

public static void received(Channel channel, Response response) { 
            try { 
            	// 等待Map移除
                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) { 
            	// 异步调用处理callback
                invokeCallback(callback);
            }
        }