Dubbo異步轉同步

　　Dubbo是一款開源的RPC中間件框架，底層數據傳輸默認使用的Netty，那么請求的處理理論上是異步的，為什么我們在使用的時候是同步的呢？肯定是Dubbo框架，做了異步轉同步的處理。

　　首先我們來梳理下，異步轉同步，我們的需求是怎樣的？

　　1、調用方請求遠程服務之后，需要等待結果，此刻，請求線程應該阻塞

　　2、遠程服務返回結果后，喚醒請求線程，調用方得到結果

　　Dubbo異步轉同步，核心類是DefaultFuture，核心方法是get()，received(Channel channel, Response response)。

　　DefaultFuture構造函數：

    private static final Map<Long, Channel> CHANNELS = new ConcurrentHashMap<Long, Channel>();

   // 每次請求都會生成一個DefaultFuture對象，然后保存到FUTURES中，請求返回結果時，根據id從FUTURES中找到對應的DefaultFuture對象，並刪除
    private static final Map<Long, DefaultFuture> FUTURES = new ConcurrentHashMap<Long, DefaultFuture>();

    // AtomicLong從0開始遞增，創建Request對象時生成的id
    private final long id;
    private final Channel channel;
    // 請求對象
    private final Request request;
    // 超時的設置
    private final int timeout;
    // 這里使用Lock和Condition實現等待通知機制
    private final Lock lock = new ReentrantLock();
    private final Condition done = lock.newCondition();
    private final long start = System.currentTimeMillis();
    private volatile long sent;
    // 請求的返回結果
    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.
        FUTURES.put(id, this);
        CHANNELS.put(id, channel);
    }

 

　　get()：

public Object get(int timeout) throws RemotingException {
        if (timeout <= 0) {
            timeout = Constants.DEFAULT_TIMEOUT;
        }
        // isDone()方法就是判斷Response是否有值(即是否有返回結果)
        if (!isDone()) {
            long start = System.currentTimeMillis();
            lock.lock();
            try {
                while (!isDone()) {
                    // 超時等待
                    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();
    }

 

　　received(Channel channel, Response response)：

public static void received(Channel channel, Response response) {
        try {
            // 根據請求id從FUTURES中獲取DefaultFuture，並刪除
            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也刪除
            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);
        }
    }

 

　　總結：Dubbo異步轉同步的原理，其實就是利用Lock和Condition實現了等待通知機制。請求與返回結果進行匹配，則是通過傳遞以及接收請求id實現的。