国产xxxx99真实实拍_久久不雅视频_高清韩国a级特黄毛片_嗯老师别我我受不了了小说

資訊專欄INFORMATION COLUMN

dubbo源碼解析(十)遠(yuǎn)程通信——Exchange層

cppprimer / 2324人閱讀

摘要:和斷開,處理措施不一樣,會分別做出重連和關(guān)閉通道的操作。取消定時器取消大量已排隊任務(wù),用于回收空間該方法是停止現(xiàn)有心跳,也就是停止定時器,釋放空間。做到異步處理返回結(jié)果時能給準(zhǔn)確的返回給對應(yīng)的請求。

遠(yuǎn)程通訊——Exchange層
目標(biāo):介紹Exchange層的相關(guān)設(shè)計和邏輯、介紹dubbo-remoting-api中的exchange包內(nèi)的源碼解析。
前言

上一篇文章我講的是dubbo框架設(shè)計中Transport層,這篇文章我要講的是它的上一層Exchange層,也就是信息交換層。官方文檔對這一層的解釋是封裝請求響應(yīng)模式,同步轉(zhuǎn)異步,以 Request, Response為中心,擴(kuò)展接口為 Exchanger, ExchangeChannel, ExchangeClient, ExchangeServer。

這一層的設(shè)計意圖是什么?它應(yīng)該算是在信息傳輸層上又做了部分裝飾,為了適應(yīng)rpc調(diào)用的一些需求,比如rpc調(diào)用中一次請求只關(guān)心它所對應(yīng)的響應(yīng),這個時候只是一個message消息傳輸過來,是無法區(qū)分這是新的請求還是上一個請求的響應(yīng),這種類似于冪等性的問題以及rpc異步處理返回結(jié)果、內(nèi)置事件等特性都是在Transport層無法解決滿足的,所有在Exchange層講message分成了request和response兩種類型,并且在這兩個模型上增加一些系統(tǒng)字段來處理問題。具體我會在下面講到。而dubbo把一條消息分為了協(xié)議頭和內(nèi)容兩部分:協(xié)議頭包括系統(tǒng)字段,例如編號等,內(nèi)容包括具體請求的參數(shù)和響應(yīng)的結(jié)果等。在exchange層中大量邏輯都是基于協(xié)議頭的。

現(xiàn)在對這一層的設(shè)計意圖大致應(yīng)該有所了解了吧,現(xiàn)在來看看exchange的類圖:

我講解的順序還是按照類圖從上而下,分塊講解,忽略綠色的test類。

源碼解析 (一)ExchangeChannel
public interface ExchangeChannel extends Channel {

    ResponseFuture request(Object request) throws RemotingException;

    ResponseFuture request(Object request, int timeout) throws RemotingException;
    
    ExchangeHandler getExchangeHandler();
    
    @Override
    void close(int timeout);

}

該接口是信息交換通道接口,有四個方法,前兩個是發(fā)送請求消息,區(qū)別就是第二個發(fā)送請求有超時的參數(shù),getExchangeHandler方法就是返回一個信息交換處理器,第四個是需要覆寫父類的方法。

(二)HeaderExchangeChannel

該類實現(xiàn)了ExchangeChannel,是基于協(xié)議頭的信息交換通道。

1.屬性
private static final Logger logger = LoggerFactory.getLogger(HeaderExchangeChannel.class);

/**
 * 通道的key值
 */
private static final String CHANNEL_KEY = HeaderExchangeChannel.class.getName() + ".CHANNEL";

/**
 * 通道
 */
private final Channel channel;

/**
 * 是否關(guān)閉
 */
private volatile boolean closed = false;

上述屬性比較簡單,還是放一下這個類的屬性是因為該類中有channel屬性,也就是說HeaderExchangeChannel是Channel的裝飾器,每個實現(xiàn)方法都會調(diào)用channel的方法。

2.靜態(tài)方法
static HeaderExchangeChannel getOrAddChannel(Channel ch) {
    if (ch == null) {
        return null;
    }
    // 獲得通道中的HeaderExchangeChannel
    HeaderExchangeChannel ret = (HeaderExchangeChannel) ch.getAttribute(CHANNEL_KEY);
    if (ret == null) {
        // 創(chuàng)建一個HeaderExchangeChannel實例
        ret = new HeaderExchangeChannel(ch);
        // 如果通道連接
        if (ch.isConnected()) {
            // 加入屬性值
            ch.setAttribute(CHANNEL_KEY, ret);
        }
    }
    return ret;
}

static void removeChannelIfDisconnected(Channel ch) {
    // 如果通道斷開連接
    if (ch != null && !ch.isConnected()) {
        // 移除屬性值
        ch.removeAttribute(CHANNEL_KEY);
    }
}

該靜態(tài)方法做了HeaderExchangeChannel的創(chuàng)建和銷毀,并且生命周期隨channel銷毀而銷毀。

3.send
@Override
public void send(Object message) throws RemotingException {
    send(message, getUrl().getParameter(Constants.SENT_KEY, false));
}

@Override
public void send(Object message, boolean sent) throws RemotingException {
    // 如果通道關(guān)閉,拋出異常
    if (closed) {
        throw new RemotingException(this.getLocalAddress(), null, "Failed to send message " + message + ", cause: The channel " + this + " is closed!");
    }
    // 判斷消息的類型
    if (message instanceof Request
            || message instanceof Response
            || message instanceof String) {
        // 發(fā)送消息
        channel.send(message, sent);
    } else {
        // 新建一個request實例
        Request request = new Request();
        // 設(shè)置信息的版本
        request.setVersion(Version.getProtocolVersion());
        // 該請求不需要響應(yīng)
        request.setTwoWay(false);
        // 把消息傳入
        request.setData(message);
        // 發(fā)送消息
        channel.send(request, sent);
    }
}

該方法是在channel的send方法上加上了request和response模型,最后再調(diào)用channel.send,起到了裝飾器的作用。

4.request
@Override
public ResponseFuture request(Object request) throws RemotingException {
    return request(request, channel.getUrl().getPositiveParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT));
}

@Override
public ResponseFuture request(Object request, int timeout) throws RemotingException {
    // 如果通道關(guān)閉,則拋出異常
    if (closed) {
        throw new RemotingException(this.getLocalAddress(), null, "Failed to send request " + request + ", cause: The channel " + this + " is closed!");
    }
    // create request.創(chuàng)建請求
    Request req = new Request();
    // 設(shè)置版本號
    req.setVersion(Version.getProtocolVersion());
    // 設(shè)置需要響應(yīng)
    req.setTwoWay(true);
    // 把請求數(shù)據(jù)傳入
    req.setData(request);
    // 創(chuàng)建DefaultFuture對象,可以從future中主動獲得請求對應(yīng)的響應(yīng)信息
    DefaultFuture future = new DefaultFuture(channel, req, timeout);
    try {
        // 發(fā)送請求消息
        channel.send(req);
    } catch (RemotingException e) {
        future.cancel();
        throw e;
    }
    return future;
}

該方法是請求方法,用Request模型把請求內(nèi)容裝飾起來,然后發(fā)送一個Request類型的消息,并且返回DefaultFuture實例,DefaultFuture我會在后面講到。

cloes方法也重寫了,我就不再多說,因為比較簡單,沒有重點,其他方法都是直接調(diào)用channel屬性的方法。

(三)ExchangeClient

該接口繼承了Client和ExchangeChannel,是信息交換客戶端接口,其中沒有定義多余的方法。

(四)HeaderExchangeClient

 該類實現(xiàn)了ExchangeClient接口,是基于協(xié)議頭的信息交互客戶端類,同樣它是Client、Channel的適配器。在該類的源碼中可以看到所有的實現(xiàn)方法都是調(diào)用了client和channel屬性的方法。該類主要的作用就是增加了心跳功能,為什么要增加心跳功能呢,對于長連接,一些拔網(wǎng)線等物理層的斷開,會導(dǎo)致TCP的FIN消息來不及發(fā)送,對方收不到斷開事件,那么就需要用到發(fā)送心跳包來檢測連接是否斷開。consumer和provider斷開,處理措施不一樣,會分別做出重連和關(guān)閉通道的操作。

1.屬性
private static final Logger logger = LoggerFactory.getLogger(HeaderExchangeClient.class);

/**
 * 定時器線程池
 */
private static final ScheduledThreadPoolExecutor scheduled = new ScheduledThreadPoolExecutor(2, new NamedThreadFactory("dubbo-remoting-client-heartbeat", true));
/**
 * 客戶端
 */
private final Client client;
/**
 * 信息交換通道
 */
private final ExchangeChannel channel;
// heartbeat timer
/**
 * 心跳定時器
 */
private ScheduledFuture heartbeatTimer;
// heartbeat(ms), default value is 0 , won"t execute a heartbeat.
/**
 * 心跳周期,間隔多久發(fā)送心跳消息檢測一次
 */
private int heartbeat;
/**
 * 心跳超時時間
 */
private int heartbeatTimeout;

該類的屬性除了需要適配的屬性外,其他都是跟心跳相關(guān)屬性。

2.構(gòu)造函數(shù)
public HeaderExchangeClient(Client client, boolean needHeartbeat) {
    if (client == null) {
        throw new IllegalArgumentException("client == null");
    }
    this.client = client;
    // 創(chuàng)建信息交換通道
    this.channel = new HeaderExchangeChannel(client);
    // 獲得dubbo版本
    String dubbo = client.getUrl().getParameter(Constants.DUBBO_VERSION_KEY);
    //獲得心跳周期配置,如果沒有配置,并且dubbo是1.0版本的,則這只為1分鐘,否則設(shè)置為0
    this.heartbeat = client.getUrl().getParameter(Constants.HEARTBEAT_KEY, dubbo != null && dubbo.startsWith("1.0.") ? Constants.DEFAULT_HEARTBEAT : 0);
    // 獲得心跳超時配置,默認(rèn)是心跳周期的三倍
    this.heartbeatTimeout = client.getUrl().getParameter(Constants.HEARTBEAT_TIMEOUT_KEY, heartbeat * 3);
    // 如果心跳超時時間小于心跳周期的兩倍,則拋出異常
    if (heartbeatTimeout < heartbeat * 2) {
        throw new IllegalStateException("heartbeatTimeout < heartbeatInterval * 2");
    }
    if (needHeartbeat) {
        // 開啟心跳
        startHeartbeatTimer();
    }
}

構(gòu)造函數(shù)就是對一些屬性初始化設(shè)置,優(yōu)先從url中獲取。心跳超時時間小于心跳周期的兩倍就拋出異常,意思就是至少重試兩次心跳檢測。

3.startHeartbeatTimer
private void startHeartbeatTimer() {
    // 停止現(xiàn)有的心跳線程
    stopHeartbeatTimer();
    // 如果需要心跳
    if (heartbeat > 0) {
        // 創(chuàng)建心跳定時器
        heartbeatTimer = scheduled.scheduleWithFixedDelay(
                // 新建一個心跳線程
                new HeartBeatTask(new HeartBeatTask.ChannelProvider() {
                    @Override
                    public Collection getChannels() {
                        // 返回一個只包含HeaderExchangeClient對象的不可變列表
                        return Collections.singletonList(HeaderExchangeClient.this);
                    }
                }, heartbeat, heartbeatTimeout),
                heartbeat, heartbeat, TimeUnit.MILLISECONDS);
    }
}

該方法就是開啟心跳。利用心跳定時器來做到定時檢測心跳。因為這是信息交換客戶端類,所有這里的只是返回包含HeaderExchangeClient對象的不可變列表,因為客戶端跟channel是一一對應(yīng)的,只有這一個該客戶端本身的channel需要心跳。

4.stopHeartbeatTimer
private void stopHeartbeatTimer() {
    if (heartbeatTimer != null && !heartbeatTimer.isCancelled()) {
        try {
            // 取消定時器
            heartbeatTimer.cancel(true);
            // 取消大量已排隊任務(wù),用于回收空間
            scheduled.purge();
        } catch (Throwable e) {
            if (logger.isWarnEnabled()) {
                logger.warn(e.getMessage(), e);
            }
        }
    }
    heartbeatTimer = null;
}

該方法是停止現(xiàn)有心跳,也就是停止定時器,釋放空間。

其他方法都是調(diào)用channel和client屬性的方法。

(五)HeartBeatTask

該類實現(xiàn)了Runnable接口,實現(xiàn)的是心跳任務(wù),里面包含了核心的心跳策略。

1.屬性
/**
 * 通道管理
 */
private ChannelProvider channelProvider;

/**
 * 心跳間隔 單位:ms
 */
private int heartbeat;

/**
 * 心跳超時時間 單位:ms
 */
private int heartbeatTimeout;

后兩個屬性跟HeaderExchangeClient中的屬性含義一樣,第一個是該類自己內(nèi)部的一個接口:

interface ChannelProvider {
    // 獲得所有的通道集合,需要心跳的通道數(shù)組
    Collection getChannels();
}

該接口就定義了一個方法,獲得需要心跳的通道集合??上攵瑫蟽?nèi)的通道都做心跳檢測。

2.run
@Override
public void run() {
    try {
        long now = System.currentTimeMillis();
        // 遍歷所有通道
        for (Channel channel : channelProvider.getChannels()) {
            // 如果通道關(guān)閉了,則跳過
            if (channel.isClosed()) {
                continue;
            }
            try {
                // 最后一次接收到消息的時間戳
                Long lastRead = (Long) channel.getAttribute(
                        HeaderExchangeHandler.KEY_READ_TIMESTAMP);
                // 最后一次發(fā)送消息的時間戳
                Long lastWrite = (Long) channel.getAttribute(
                        HeaderExchangeHandler.KEY_WRITE_TIMESTAMP);
                // 如果最后一次接收或者發(fā)送消息到時間到現(xiàn)在的時間間隔超過了心跳間隔時間
                if ((lastRead != null && now - lastRead > heartbeat)
                        || (lastWrite != null && now - lastWrite > heartbeat)) {
                    // 創(chuàng)建一個request
                    Request req = new Request();
                    // 設(shè)置版本號
                    req.setVersion(Version.getProtocolVersion());
                    // 設(shè)置需要得到響應(yīng)
                    req.setTwoWay(true);
                    // 設(shè)置事件類型,為心跳事件
                    req.setEvent(Request.HEARTBEAT_EVENT);
                    // 發(fā)送心跳請求
                    channel.send(req);
                    if (logger.isDebugEnabled()) {
                        logger.debug("Send heartbeat to remote channel " + channel.getRemoteAddress()
                                + ", cause: The channel has no data-transmission exceeds a heartbeat period: " + heartbeat + "ms");
                    }
                }
                // 如果最后一次接收消息的時間到現(xiàn)在已經(jīng)超過了超時時間
                if (lastRead != null && now - lastRead > heartbeatTimeout) {
                    logger.warn("Close channel " + channel
                            + ", because heartbeat read idle time out: " + heartbeatTimeout + "ms");
                    // 如果該通道是客戶端,也就是請求的服務(wù)器掛掉了,客戶端嘗試重連服務(wù)器
                    if (channel instanceof Client) {
                        try {
                            // 重新連接服務(wù)器
                            ((Client) channel).reconnect();
                        } catch (Exception e) {
                            //do nothing
                        }
                    } else {
                        // 如果不是客戶端,也就是是服務(wù)端返回響應(yīng)給客戶端,但是客戶端掛掉了,則服務(wù)端關(guān)閉客戶端連接
                        channel.close();
                    }
                }
            } catch (Throwable t) {
                logger.warn("Exception when heartbeat to remote channel " + channel.getRemoteAddress(), t);
            }
        }
    } catch (Throwable t) {
        logger.warn("Unhandled exception when heartbeat, cause: " + t.getMessage(), t);
    }
}

該方法中是心跳機(jī)制的核心邏輯。注意以下幾個點:

如果需要心跳的通道本身如果關(guān)閉了,那么跳過,不添加心跳機(jī)制。

無論是接收消息還是發(fā)送消息,只要超過了設(shè)置的心跳間隔,就發(fā)送心跳消息來測試是否斷開

如果最后一次接收到消息到到現(xiàn)在已經(jīng)超過了心跳超時時間,那就認(rèn)定對方的確斷開,分兩種情況來處理對方斷開的情況。分別是服務(wù)端斷開,客戶端重連以及客戶端斷開,服務(wù)端斷開這個客戶端的連接。,這里要好好品味一下誰是發(fā)送方,誰在等誰的響應(yīng),苦苦沒有等到。

(六)ResponseFuture
public interface ResponseFuture {

    Object get() throws RemotingException;

    Object get(int timeoutInMillis) throws RemotingException;

    void setCallback(ResponseCallback callback);

    boolean isDone();

}

該接口是響應(yīng)future接口,該接口的設(shè)計意圖跟java.util.concurrent.Future很類似。發(fā)送出去的消息,潑出去的水,只有等到對方主動響應(yīng)才能得到結(jié)果,但是請求方需要去主動回去該請求的結(jié)果,就顯得有些艱難,所有產(chǎn)生了這樣一個接口,它能夠獲取任務(wù)執(zhí)行結(jié)果、可以核對請求消息是否被響應(yīng),還能設(shè)置回調(diào)來支持異步。

(七)DefaultFuture

該類實現(xiàn)了ResponseFuture接口,其中封裝了處理響應(yīng)的邏輯。你可以把DefaultFuture看成是一個中介,買房和賣房都通過這個中介進(jìn)行溝通,中介擁有著買房者的信息request和賣房者的信息response,并且促成他們之間的買賣。

1.屬性
private static final Logger logger = LoggerFactory.getLogger(DefaultFuture.class);

/**
 * 通道集合
 */
private static final Map CHANNELS = new ConcurrentHashMap();

/**
 * Future集合,key為請求編號
 */
private static final Map FUTURES = new ConcurrentHashMap();

// invoke 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();
/**
 * 創(chuàng)建開始時間
 */
private final long start = System.currentTimeMillis();
/**
 * 發(fā)送請求時間
 */
private volatile long sent;
/**
 * 響應(yīng)
 */
private volatile Response response;
/**
 * 回調(diào)
 */
private volatile ResponseCallback callback;

可以看到,該類的屬性包含了request、response、channel三個實例,在該類中,把請求和響應(yīng)通過唯一的id一一對應(yīng)起來。做到異步處理返回結(jié)果時能給準(zhǔn)確的返回給對應(yīng)的請求。可以看到屬性中有兩個集合,分別是通道集合和future集合,也就是該類本身也是所有 DefaultFuture 的管理容器。

2.構(gòu)造函數(shù)
public DefaultFuture(Channel channel, Request request, int timeout) {
    this.channel = channel;
    this.request = request;
    // 設(shè)置請求編號
    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);
}

構(gòu)造函數(shù)比較簡單,每一個DefaultFuture實例都跟每一個請求一一對應(yīng),被存入到集合中管理起來。

3.closeChannel
public static void closeChannel(Channel channel) {
    // 遍歷通道集合
    for (long id : CHANNELS.keySet()) {
        if (channel.equals(CHANNELS.get(id))) {
            // 通過請求id獲得future
            DefaultFuture future = getFuture(id);
            if (future != null && !future.isDone()) {
                // 創(chuàng)建一個關(guān)閉通道的響應(yīng)
                Response disconnectResponse = new Response(future.getId());
                disconnectResponse.setStatus(Response.CHANNEL_INACTIVE);
                disconnectResponse.setErrorMessage("Channel " +
                        channel +
                        " is inactive. Directly return the unFinished request : " +
                        future.getRequest());
                // 接收該關(guān)閉通道并且請求未完成的響應(yīng)
                DefaultFuture.received(channel, disconnectResponse);
            }
        }
    }
}

該方法是關(guān)閉不活躍的通道,并且返回請求未完成。也就是關(guān)閉指定channel的請求,返回的是請求未完成。

4.received
public static void received(Channel channel, Response response) {
    try {
        // future集合中移除該請求的future,(響應(yīng)id和請求id一一對應(yīng)的)
        DefaultFuture future = FUTURES.remove(response.getId());
        if (future != null) {
            // 接收響應(yīng)結(jié)果
            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 {
        // 通道集合移除該請求對應(yīng)的通道,代表著這一次請求結(jié)束
        CHANNELS.remove(response.getId());
    }
}

該方法是接收響應(yīng),也就是某個請求得到了響應(yīng),那么代表這次請求任務(wù)完成,所有需要把future從集合中移除。具體的接收響應(yīng)結(jié)果在doReceived方法中實現(xiàn)。

5.doReceived
private void doReceived(Response res) {
    // 獲得鎖
    lock.lock();
    try {
        // 設(shè)置響應(yīng)
        response = res;
        if (done != null) {
            // 喚醒等待
            done.signal();
        }
    } finally {
        // 釋放鎖
        lock.unlock();
    }
    if (callback != null) {
        // 執(zhí)行回調(diào)
        invokeCallback(callback);
    }
}

可以看到,當(dāng)接收到響應(yīng)后,會把等待的線程喚醒,然后執(zhí)行回調(diào)來處理該響應(yīng)結(jié)果。

6.invokeCallback
private void invokeCallback(ResponseCallback c) {
    ResponseCallback callbackCopy = c;
    if (callbackCopy == null) {
        throw new NullPointerException("callback cannot be null.");
    }
    c = null;
    Response res = response;
    if (res == null) {
        throw new IllegalStateException("response cannot be null. url:" + channel.getUrl());
    }

    // 如果響應(yīng)成功,返回碼是20
    if (res.getStatus() == Response.OK) {
        try {
            // 使用響應(yīng)結(jié)果執(zhí)行 完成 后的邏輯
            callbackCopy.done(res.getResult());
        } catch (Exception e) {
            logger.error("callback invoke error .reasult:" + res.getResult() + ",url:" + channel.getUrl(), e);
        }
        //超時,回調(diào)處理成超時異常
    } else if (res.getStatus() == Response.CLIENT_TIMEOUT || res.getStatus() == Response.SERVER_TIMEOUT) {
        try {
            TimeoutException te = new TimeoutException(res.getStatus() == Response.SERVER_TIMEOUT, channel, res.getErrorMessage());
            // 回調(diào)處理異常
            callbackCopy.caught(te);
        } catch (Exception e) {
            logger.error("callback invoke error ,url:" + channel.getUrl(), e);
        }
        // 其他情況處理成RemotingException異常
    } else {
        try {
            RuntimeException re = new RuntimeException(res.getErrorMessage());
            callbackCopy.caught(re);
        } catch (Exception e) {
            logger.error("callback invoke error ,url:" + channel.getUrl(), e);
        }
    }
}

該方法是執(zhí)行回調(diào)來處理響應(yīng)結(jié)果。分為了三種情況:

響應(yīng)成功,那么執(zhí)行完成后的邏輯。

超時,會按照超時異常來處理

其他,按照RuntimeException異常來處理

具體的處理都在ResponseCallback接口的實現(xiàn)類里執(zhí)行,后面我會講到。

7.get
@Override
public Object get() throws RemotingException {
    return get(timeout);
}

@Override
public Object get(int timeout) throws RemotingException {
    // 超時時間默認(rèn)為1s
    if (timeout <= 0) {
        timeout = Constants.DEFAULT_TIMEOUT;
    }
    // 如果請求沒有完成,也就是還沒有響應(yīng)返回
    if (!isDone()) {
        long start = System.currentTimeMillis();
        // 獲得鎖
        lock.lock();
        try {
            // 輪詢 等待請求是否完成
            while (!isDone()) {
                // 線程阻塞等待
                done.await(timeout, TimeUnit.MILLISECONDS);
                // 如果請求完成或者超時,則結(jié)束
                if (isDone() || System.currentTimeMillis() - start > timeout) {
                    break;
                }
            }
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } finally {
            // 釋放鎖
            lock.unlock();
        }
        // 如果沒有收到響應(yīng),則拋出超時的異常
        if (!isDone()) {
            throw new TimeoutException(sent > 0, channel, getTimeoutMessage(false));
        }
    }
    // 返回響應(yīng)
    return returnFromResponse();
}

該方法是實現(xiàn)了ResponseFuture定義的方法,是獲得該future對應(yīng)的請求對應(yīng)的響應(yīng)結(jié)果,其實future、請求、響應(yīng)都是一一對應(yīng)的。其中如果還沒得到響應(yīng),則會線程阻塞等待,等到有響應(yīng)結(jié)果或者超時,才返回。返回的邏輯在returnFromResponse中實現(xiàn)。

8.returnFromResponse
private Object returnFromResponse() throws RemotingException {
    Response res = response;
    if (res == null) {
        throw new IllegalStateException("response cannot be null");
    }
    // 如果正常返回,則返回響應(yīng)結(jié)果
    if (res.getStatus() == Response.OK) {
        return res.getResult();
    }
    // 如果超時,則拋出超時異常
    if (res.getStatus() == Response.CLIENT_TIMEOUT || res.getStatus() == Response.SERVER_TIMEOUT) {
        throw new TimeoutException(res.getStatus() == Response.SERVER_TIMEOUT, channel, res.getErrorMessage());
    }
    // 其他 拋出RemotingException異常
    throw new RemotingException(channel, res.getErrorMessage());
}

這代碼跟invokeCallback方法中差不多,都是把響應(yīng)分了三種情況。

9.cancel
public void cancel() {
    // 創(chuàng)建一個取消請求的響應(yīng)
    Response errorResult = new Response(id);
    errorResult.setErrorMessage("request future has been canceled.");
    response = errorResult;
    // 從集合中刪除該請求
    FUTURES.remove(id);
    CHANNELS.remove(id);
}

該方法是取消一個請求,可以直接關(guān)閉一個請求,也就是值創(chuàng)建一個響應(yīng)來回應(yīng)該請求,把response值設(shè)置到該請求對于到future中,做到了中斷請求的作用。該方法跟closeChannel的區(qū)別是closeChannel中對response的狀態(tài)設(shè)置了CHANNEL_INACTIVE,而cancel方法是中途被主動取消的,雖然有response值,但是并沒有一個響應(yīng)狀態(tài)。

10.RemotingInvocationTimeoutScan
private static class RemotingInvocationTimeoutScan implements Runnable {

    @Override
    public void run() {
        while (true) {
            try {
                for (DefaultFuture future : FUTURES.values()) {
                    // 已經(jīng)完成,跳過掃描
                    if (future == null || future.isDone()) {
                        continue;
                    }
                    // 超時
                    if (System.currentTimeMillis() - future.getStartTimestamp() > future.getTimeout()) {
                        // create exception response.,創(chuàng)建一個超時的響應(yīng)
                        Response timeoutResponse = new Response(future.getId());
                        // set timeout status.,設(shè)置超時狀態(tài),是服務(wù)端側(cè)超時還是客戶端側(cè)超時
                        timeoutResponse.setStatus(future.isSent() ? Response.SERVER_TIMEOUT : Response.CLIENT_TIMEOUT);
                        // 設(shè)置錯誤信息
                        timeoutResponse.setErrorMessage(future.getTimeoutMessage(true));
                        // handle response.,接收創(chuàng)建的超時響應(yīng)
                        DefaultFuture.received(future.getChannel(), timeoutResponse);
                    }
                }
                // 睡眠
                Thread.sleep(30);
            } catch (Throwable e) {
                logger.error("Exception when scan the timeout invocation of remoting.", e);
            }
        }
    }
}

該方法是掃描調(diào)用超時任務(wù)的線程,每次都會遍歷future集合,檢測請求是否超時了,如果超時則創(chuàng)建一個超時響應(yīng)來回應(yīng)該請求。

static {
    // 開啟一個后臺掃描調(diào)用超時任務(wù)
    Thread th = new Thread(new RemotingInvocationTimeoutScan(), "DubboResponseTimeoutScanTimer");
    th.setDaemon(true);
    th.start();
}

開啟一個后臺線程進(jìn)行掃描的邏輯寫在了靜態(tài)代碼塊里面,只開啟一次。

(八)SimpleFuture

該類實現(xiàn)了ResponseFuture,目前沒有用到,很簡單的實現(xiàn),我就不多說了。

(九)ExchangeHandler

該接口繼承了ChannelHandler, TelnetHandler接口,是信息交換處理器接口。

public interface ExchangeHandler extends ChannelHandler, TelnetHandler {
    /**
     * reply.
     * 回復(fù)請求結(jié)果
     * @param channel
     * @param request
     * @return response
     * @throws RemotingException
     */
    Object reply(ExchangeChannel channel, Object request) throws RemotingException;

}

該接口只定義了一個回復(fù)請求結(jié)果的方法,返回的是請求結(jié)果。

(十)ExchangeHandlerDispatcher

該類實現(xiàn)了ExchangeHandler接口, 是信息交換處理器調(diào)度器類,也就是對應(yīng)不同的事件,選擇不同的處理器去處理。該類中有三個屬性,分別對應(yīng)了三種事件:

/**
 * 回復(fù)者調(diào)度器
 */
private final ReplierDispatcher replierDispatcher;

/**
 * 通道處理器調(diào)度器
 */
private final ChannelHandlerDispatcher handlerDispatcher;

/**
 * Telnet 命令處理器
 */
private final TelnetHandler telnetHandler;

如果事件是跟通道處理器有關(guān)的,就調(diào)用通道處理器來處理,比如:

@Override
@SuppressWarnings({"unchecked", "rawtypes"})
public Object reply(ExchangeChannel channel, Object request) throws RemotingException {
    return ((Replier) replierDispatcher).reply(channel, request);
}

@Override
public void connected(Channel channel) {
    handlerDispatcher.connected(channel);
}
@Override
public String telnet(Channel channel, String message) throws RemotingException {
    return telnetHandler.telnet(channel, message);
}

可以看到以上三種事件,回復(fù)請求結(jié)果需要回復(fù)者調(diào)度器來處理,連接需要通道處理器調(diào)度器來處理,telnet消息需要Telnet命令處理器來處理。

(十一)ExchangeHandlerAdapter

該類繼承了TelnetHandlerAdapter,實現(xiàn)了ExchangeHandler,是信息交換處理器的適配器類。

public abstract class ExchangeHandlerAdapter extends TelnetHandlerAdapter implements ExchangeHandler {

    @Override
    public Object reply(ExchangeChannel channel, Object msg) throws RemotingException {
        // 直接返回null
        return null;
    }

}

該類直接讓ExchangeHandler定義的方法reply返回null,交由它的子類選擇性的去實現(xiàn)具體的回復(fù)請求結(jié)果。

(十二)ExchangeServer

該接口繼承了Server接口,定義了兩個方法:

public interface ExchangeServer extends Server {

    /**
     * get channels.
     * 獲得通道集合
     * @return channels
     */
    Collection getExchangeChannels();

    /**
     * get channel.
     * 根據(jù)遠(yuǎn)程地址獲得對應(yīng)的信息通道
     * @param remoteAddress
     * @return channel
     */
    ExchangeChannel getExchangeChannel(InetSocketAddress remoteAddress);

}

該接口比較好理解,并且在Server接口基礎(chǔ)上新定義了兩個方法。直接來看看它的實現(xiàn)類吧。

(十三)HeaderExchangeServer

該類實現(xiàn)了ExchangeServer接口,是基于協(xié)議頭的信息交換服務(wù)器實現(xiàn)類,HeaderExchangeServer是Server的裝飾器,每個實現(xiàn)方法都會調(diào)用server的方法。

1.屬性
protected final Logger logger = LoggerFactory.getLogger(getClass());

/**
 * 線程池
 */
private final ScheduledExecutorService scheduled = Executors.newScheduledThreadPool(1,
        new NamedThreadFactory(
                "dubbo-remoting-server-heartbeat",
                true));
/**
 * 服務(wù)器
 */
private final Server server;
// heartbeat timer
/**
 * 心跳定時器
 */
private ScheduledFuture heartbeatTimer;
// heartbeat timeout (ms), default value is 0 , won"t execute a heartbeat.
/**
 * 心跳周期
 */
private int heartbeat;
/**
 * 心跳超時時間
 */
private int heartbeatTimeout;
/**
 * 信息交換服務(wù)器是否關(guān)閉
 */
private AtomicBoolean closed = new AtomicBoolean(false);

該類里面的很多實現(xiàn)跟HeaderExchangeClient差不多,包括心跳檢測等邏輯??吹枚鲜鑫抑v的HeaderExchangeClient的屬性,想必這里的屬性應(yīng)該也很簡單了。

2.構(gòu)造函數(shù)
public HeaderExchangeServer(Server server) {
    if (server == null) {
        throw new IllegalArgumentException("server == null");
    }
    this.server = server;
    //獲得心跳周期配置,如果沒有配置,默認(rèn)設(shè)置為0
    this.heartbeat = server.getUrl().getParameter(Constants.HEARTBEAT_KEY, 0);
    // 獲得心跳超時配置,默認(rèn)是心跳周期的三倍
    this.heartbeatTimeout = server.getUrl().getParameter(Constants.HEARTBEAT_TIMEOUT_KEY, heartbeat * 3);
    // 如果心跳超時時間小于心跳周期的兩倍,則拋出異常
    if (heartbeatTimeout < heartbeat * 2) {
        throw new IllegalStateException("heartbeatTimeout < heartbeatInterval * 2");
    }
    // 開始心跳
    startHeartbeatTimer();
}

public Server getServer() {
    return server;
}

構(gòu)造函數(shù)就是對屬性的設(shè)置,心跳的機(jī)制以及默認(rèn)值都跟HeaderExchangeClient中的一模一樣。

3.isRunning
private boolean isRunning() {
    Collection channels = getChannels();
    // 遍歷所有連接該服務(wù)器的通道
    for (Channel channel : channels) {

        /**
         *  If there are any client connections,
         *  our server should be running.
         */

        // 只要有任何一個客戶端連接,則服務(wù)器還運(yùn)行著
        if (channel.isConnected()) {
            return true;
        }
    }
    return false;
}

該方法是檢測服務(wù)器是否還運(yùn)行,只要有一個客戶端連接著,就算服務(wù)器運(yùn)行著。

4.close
@Override
public void close() {
    // 關(guān)閉線程池和心跳檢測
    doClose();
    // 關(guān)閉服務(wù)器
    server.close();
}

@Override
public void close(final int timeout) {
    // 開始關(guān)閉
    startClose();
    if (timeout > 0) {
        final long max = (long) timeout;
        final long start = System.currentTimeMillis();
        if (getUrl().getParameter(Constants.CHANNEL_SEND_READONLYEVENT_KEY, true)) {
            // 發(fā)送 READONLY_EVENT事件給所有連接該服務(wù)器的客戶端,表示 Server 不可讀了。
            sendChannelReadOnlyEvent();
        }
        // 當(dāng)服務(wù)器還在運(yùn)行,并且沒有超時,睡眠,也就是等待timeout左右時間在進(jìn)行關(guān)閉
        while (HeaderExchangeServer.this.isRunning()
                && System.currentTimeMillis() - start < max) {
            try {
                Thread.sleep(10);
            } catch (InterruptedException e) {
                logger.warn(e.getMessage(), e);
            }
        }
    }
    // 關(guān)閉線程池和心跳檢測
    doClose();
    // 延遲關(guān)閉
    server.close(timeout);
}

兩個close方法,第二個close方法是優(yōu)雅的關(guān)閉,有一定的延時來讓一些響應(yīng)或者操作做完。關(guān)閉分兩個步驟,第一個就是關(guān)閉信息交換服務(wù)器中的線程池和心跳檢測,然后才是關(guān)閉服務(wù)器。

5.sendChannelReadOnlyEvent
private void sendChannelReadOnlyEvent() {
    // 創(chuàng)建一個READONLY_EVENT事件的請求
    Request request = new Request();
    request.setEvent(Request.READONLY_EVENT);
    // 不需要響應(yīng)
    request.setTwoWay(false);
    // 設(shè)置版本
    request.setVersion(Version.getProtocolVersion());

    Collection channels = getChannels();
    // 遍歷連接的通道,進(jìn)行通知
    for (Channel channel : channels) {
        try {
            // 通過通道還連接著,則發(fā)送通知
            if (channel.isConnected())
                channel.send(request, getUrl().getParameter(Constants.CHANNEL_READONLYEVENT_SENT_KEY, true));
        } catch (RemotingException e) {
            logger.warn("send cannot write message error.", e);
        }
    }
}

在關(guān)閉服務(wù)器中有一個操作就是發(fā)送事件READONLY_EVENT,告訴客戶端該服務(wù)器不可讀了,就是該方法實現(xiàn)的,逐個通知連接的客戶端該事件。

6.doClose
private void doClose() {
    if (!closed.compareAndSet(false, true)) {
        return;
    }
    // 停止心跳檢測
    stopHeartbeatTimer();
    try {
        // 關(guān)閉線程池
        scheduled.shutdown();
    } catch (Throwable t) {
        logger.warn(t.getMessage(), t);
    }
}

該方法就是close方法調(diào)用到的停止心跳檢測和關(guān)閉線程池。

7.getExchangeChannels
@Override
public Collection getExchangeChannels() {
    Collection exchangeChannels = new ArrayList();
    // 獲得連接該服務(wù)器通道集合
    Collection channels = server.getChannels();
    if (channels != null && !channels.isEmpty()) {
        // 遍歷通道集合,為每個通道都創(chuàng)建信息交換通道,并且加入信息交換通道集合
        for (Channel channel : channels) {
            exchangeChannels.add(HeaderExchangeChannel.getOrAddChannel(channel));
        }
    }
    return exchangeChannels;
}

該方法是返回連接該服務(wù)器信息交換通道集合。邏輯就是先獲得通道集合,在根據(jù)通道來創(chuàng)建信息交換通道,然后返回信息通道集合。

8.reset
@Override
public void reset(URL url) {
    // 重置屬性
    server.reset(url);
    try {
        // 重置的邏輯跟構(gòu)造函數(shù)一樣設(shè)置
        if (url.hasParameter(Constants.HEARTBEAT_KEY)
                || url.hasParameter(Constants.HEARTBEAT_TIMEOUT_KEY)) {
            int h = url.getParameter(Constants.HEARTBEAT_KEY, heartbeat);
            int t = url.getParameter(Constants.HEARTBEAT_TIMEOUT_KEY, h * 3);
            if (t < h * 2) {
                throw new IllegalStateException("heartbeatTimeout < heartbeatInterval * 2");
            }
            if (h != heartbeat || t != heartbeatTimeout) {
                heartbeat = h;
                heartbeatTimeout = t;
                // 重新開始心跳
                startHeartbeatTimer();
            }
        }
    } catch (Throwable t) {
        logger.error(t.getMessage(), t);
    }
}

該方法就是重置屬性,重置后,重新開始心跳,設(shè)置心跳屬性的機(jī)制跟構(gòu)造函數(shù)一樣。

9.startHeartbeatTimer
private void startHeartbeatTimer() {
    // 先停止現(xiàn)有的心跳檢測
    stopHeartbeatTimer();
    if (heartbeat > 0) {
        // 創(chuàng)建心跳定時器
        heartbeatTimer = scheduled.scheduleWithFixedDelay(
                new HeartBeatTask(new HeartBeatTask.ChannelProvider() {
                    @Override
                    public Collection getChannels() {
                        // 返回一個不可修改的連接該服務(wù)器的信息交換通道集合
                        return Collections.unmodifiableCollection(
                                HeaderExchangeServer.this.getChannels());
                    }
                }, heartbeat, heartbeatTimeout),
                heartbeat, heartbeat, TimeUnit.MILLISECONDS);
    }
}

該方法是開始心跳,跟HeaderExchangeClient類中的開始心跳方法唯一區(qū)別是獲得的通道不一樣,客戶端跟通道是一一對應(yīng)的,所有只要對一個通道進(jìn)行心跳檢測,而服務(wù)端跟通道是一對多的關(guān)系,所有需要對該服務(wù)器連接的所有通道進(jìn)行心跳檢測。

10.stopHeartbeatTimer
private void stopHeartbeatTimer() {
    if (heartbeatTimer != null && !heartbeatTimer.isCancelled()) {
        try {
            // 取消定時器
            heartbeatTimer.cancel(true);
            // 取消大量已排隊任務(wù),用于回收空間
            scheduled.purge();
        } catch (Throwable e) {
            if (logger.isWarnEnabled()) {
                logger.warn(e.getMessage(), e);
            }
        }
    }
    heartbeatTimer = null;
}

該方法是停止當(dāng)前的心跳檢測。

(十四)ExchangeServerDelegate

該類實現(xiàn)了ExchangeServer接口,是信息交換服務(wù)器裝飾者,是ExchangeServer的裝飾器。該類就一個屬性ExchangeServer server,所有實現(xiàn)方法都調(diào)用了server屬性的方法。目前只有在p2p中被用到,代碼為就不貼了,很簡單。

(十五)Exchanger
@SPI(HeaderExchanger.NAME)
public interface Exchanger {

    /**
     * bind.
     * 綁定一個服務(wù)器
     * @param url 服務(wù)器url
     * @param handler 數(shù)據(jù)交換處理器
     * @return message server 數(shù)據(jù)交換服務(wù)器
     */
    @Adaptive({Constants.EXCHANGER_KEY})
    ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException;

    /**
     * connect.
     * 連接一個服務(wù)器,也就是創(chuàng)建一個客戶端
     * @param url 服務(wù)器url
     * @param handler 數(shù)據(jù)交換處理器
     * @return message channel 返回數(shù)據(jù)交換客戶端
     */
    @Adaptive({Constants.EXCHANGER_KEY})
    ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException;

}

該接口是數(shù)據(jù)交換者接口,該接口是一個可擴(kuò)展接口默認(rèn)實現(xiàn)的是HeaderExchanger類,并且用到了dubbo SPI的Adaptive機(jī)制,優(yōu)先實現(xiàn)url攜帶的配置。如果不了解dubbo SPI機(jī)制的可以看《dubbo源碼解析(二)Dubbo擴(kuò)展機(jī)制SPI》。那么回到該接口定義的方法,定義了綁定和連接兩個方法,分別返回信息交互服務(wù)器和客戶端實例。

(十六)HeaderExchanger
public class HeaderExchanger implements Exchanger {

    public static final String NAME = "header";

    @Override
    public ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException {
        // 用傳輸層連接返回的client 創(chuàng)建對應(yīng)的信息交換客戶端,默認(rèn)開啟心跳檢測
        return new HeaderExchangeClient(Transporters.connect(url, new DecodeHandler(new HeaderExchangeHandler(handler))), true);
    }

    @Override
    public ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException {
        // 用傳輸層綁定返回的server 創(chuàng)建對應(yīng)的信息交換服務(wù)端
        return new HeaderExchangeServer(Transporters.bind(url, new DecodeHandler(new HeaderExchangeHandler(handler))));
    }

}

該類繼承了Exchanger接口,是Exchanger接口的默認(rèn)實現(xiàn),實現(xiàn)了Exchanger接口定義的兩個方法,分別調(diào)用的是Transporters的連接和綁定方法,再利用這這兩個方法返回的客戶端和服務(wù)端實例來創(chuàng)建信息交換的客戶端和服務(wù)端。

(十七)Replier

我們知道Request對應(yīng)的是ExchangeHandler接口實現(xiàn)對象來處理,但有些時候我們需要不同數(shù)據(jù)類型對應(yīng)不同的處理器,該類就是為了支持這一需求所設(shè)計的。

public interface Replier {

    /**
     * reply.
     * 回復(fù)請求結(jié)果
     * @param channel
     * @param request
     * @return response
     * @throws RemotingException
     */
    Object reply(ExchangeChannel channel, T request) throws RemotingException;

}

可以看到該接口跟ExchangeHandler定義的方法也一一,只有請求的類型改為了范型。

(十八)ReplierDispatcher

該類實現(xiàn)了Replier接口,是回復(fù)者調(diào)度器實現(xiàn)類。

/**
 * 默認(rèn)回復(fù)者
 */
private final Replier defaultReplier;

/**
 * 回復(fù)者集合
 */
private final Map, Replier> repliers = new ConcurrentHashMap, Replier>();

這是該類的兩個屬性,緩存了回復(fù)者集合和默認(rèn)的回復(fù)者。

/**
 * 從回復(fù)者集合中找到該類型的回復(fù)者,并且返回
 * @param type
 * @return
 */
private Replier getReplier(Class type) {
    for (Map.Entry, Replier> entry : repliers.entrySet()) {
        if (entry.getKey().isAssignableFrom(type)) {
            return entry.getValue();
        }
    }
    if (defaultReplier != null) {
        return defaultReplier;
    }
    throw new IllegalStateException("Replier not found, Unsupported message object: " + type);
}

/**
 * 回復(fù)請求
 * @param channel
 * @param request
 * @return
 * @throws RemotingException
 */
@Override
@SuppressWarnings({"unchecked", "rawtypes"})
public Object reply(ExchangeChannel channel, Object request) throws RemotingException {
    return ((Replier) getReplier(request.getClass())).reply(channel, request);
}

上述是該類中關(guān)鍵的兩個方法,reply還是調(diào)用實現(xiàn)類的reply。根據(jù)請求的數(shù)據(jù)類型來使用指定的回復(fù)者進(jìn)行回復(fù)。

(十九)MultiMessage

該類實現(xiàn)了實現(xiàn) Iterable 接口,是多消息的封裝,我們直接看它的屬性:

/**
 * 消息集合
 */
private final List messages = new ArrayList();

該類要和《dubbo源碼解析(九)遠(yuǎn)程通信——Transport層》的(八)MultiMessageHandler聯(lián)合著看。

(二十)HeartbeatHandler

該類繼承了AbstractChannelHandlerDelegate類,是心跳處理器。是用來處理心跳事件的,也接收消息上增加了對心跳消息的處理。該類是

@Override
public void received(Channel channel, Object message) throws RemotingException {
    // 設(shè)置接收時間的時間戳屬性值
    setReadTimestamp(channel);
    // 如果是心跳請求
    if (isHeartbeatRequest(message)) {
        Request req = (Request) message;
        // 如果需要響應(yīng)
        if (req.isTwoWay()) {
            // 創(chuàng)建一個響應(yīng)
            Response res = new Response(req.getId(), req.getVersion());
            // 設(shè)置為心跳事件的響應(yīng)
            res.setEvent(Response.HEARTBEAT_EVENT);
            // 發(fā)送消息,也就是返回響應(yīng)
            channel.send(res);
            if (logger.isInfoEnabled()) {
                int heartbeat = channel.getUrl().getParameter(Constants.HEARTBEAT_KEY, 0);
                if (logger.isDebugEnabled()) {
                    logger.debug("Received heartbeat from remote channel " + channel.getRemoteAddress()
                            + ", cause: The channel has no data-transmission exceeds a heartbeat period"
                            + (heartbeat > 0 ? ": " + heartbeat + "ms" : ""));
                }
            }
        }
        return;
    }
    // 如果是心跳響應(yīng),則直接return
    if (isHeartbeatResponse(message)) {
        if (logger.isDebugEnabled()) {
            logger.debug("Receive heartbeat response in thread " + Thread.currentThread().getName());
        }
        return;
    }
    handler.received(channel, message);
}

該方法是就是在handler處理消息上增加了處理心跳消息的功能,做到了功能增強(qiáng)。

(二十一)Exchangers

該類跟Transporters的設(shè)計意圖是一樣的,Transporters我在《dubbo源碼解析(八)遠(yuǎn)程通信——開篇》的(十)Transporters已經(jīng)講到了。Exchangers也用到了外觀模式。代碼為就不貼了,可以對照著Transporters來看,很簡單。

(二十二)Request

請求模型類,最重要的肯定是模型的屬性,我們來看看屬性:

/**
 * 心跳事件
 */
public static final String HEARTBEAT_EVENT = null;

/**
 * 只讀事件
 */
public static final String READONLY_EVENT = "R";

/**
 * 請求編號自增序列
 */
private static final AtomicLong INVOKE_ID = new AtomicLong(0);

/**
 * 請求編號
 */
private final long mId;

/**
 * dubbo版本
 */
private String mVersion;

/**
 * 是否需要響應(yīng)
 */
private boolean mTwoWay = true;

/**
 * 是否是事件
 */
private boolean mEvent = false;

/**
 * 是否是異常的請求
 */
private boolean mBroken = false;

/**
 * 請求數(shù)據(jù)
 */
private Object mData;

由于心跳事件比較常用,所有設(shè)置為null。

請求編號使用INVOKE_ID生成,是JVM 進(jìn)程內(nèi)唯一的。

其他屬性比較簡單

(二十三)Response

響應(yīng)模型,來看看它的屬性:

/**
 * 心跳事件
 */
public static final String HEARTBEAT_EVENT = null;

/**
 * 只讀事件
 */
public static final String READONLY_EVENT = "R";

/**
 * ok.
 * 成功狀態(tài)碼
 */
public static final byte OK = 20;

/**
 * clien side timeout.
 * 客戶端側(cè)的超時狀態(tài)碼
 */
public static final byte CLIENT_TIMEOUT = 30;

/**
 * server side timeout.
 * 服務(wù)端側(cè)超時的狀態(tài)碼
 */
public static final byte SERVER_TIMEOUT = 31;

/**
 * channel inactive, directly return the unfinished requests.
 * 通道不活躍,返回未完成請求的狀態(tài)碼
 */
public static final byte CHANNEL_INACTIVE = 35;

/**
 * request format error.
 * 請求格式錯誤狀態(tài)碼
 */
public static final byte BAD_REQUEST = 40;

/**
 * response format error.
 * 響應(yīng)格式錯誤狀態(tài)碼
 */
public static final byte BAD_RESPONSE = 50;

/**
 * service not found.
 * 服務(wù)找不到狀態(tài)碼
 */
public static final byte SERVICE_NOT_FOUND = 60;

/**
 * service error.
 * 服務(wù)錯誤狀態(tài)碼
 */
public static final byte SERVICE_ERROR = 70;

/**
 * internal server error.
 * 內(nèi)部服務(wù)器錯誤狀態(tài)碼
 */
public static final byte SERVER_ERROR = 80;

/**
 * internal server error.
 * 客戶端錯誤狀態(tài)碼
 */
public static final byte CLIENT_ERROR = 90;

/**
 * server side threadpool exhausted and quick return.
 * 服務(wù)器端線程池耗盡并快速返回狀態(tài)碼
 */
public static final byte SERVER_THREADPOOL_EXHAUSTED_ERROR = 100;

/**
 * 響應(yīng)編號
 */
private long mId = 0;

/**
 * dubbo 版本
 */
private String mVersion;

/**
 * 狀態(tài)
 */
private byte mStatus = OK;

/**
 * 是否是事件
 */
private boolean mEvent = false;

/**
 * 錯誤信息
 */
private String mErrorMsg;

/**
 * 返回結(jié)果
 */
private Object mResult;

很多屬性跟Request模型的屬性一樣,并且含義也一樣,不過該模型多了很多的狀態(tài)碼。關(guān)鍵的是id跟請求一一對應(yīng)。

(二十四)ResponseCallback
public interface ResponseCallback {

    /**
     * done.
     * 處理請求
     * @param response
     */
    void done(Object response);

    /**
     * caught exception.
     * 處理異常
     * @param exception
     */
    void caught(Throwable exception);

}

該接口是回調(diào)的接口,定義了兩個方法,分別是處理正常的響應(yīng)結(jié)果和處理異常。

(二十五)ExchangeCodec

該類繼承了TelnetCodec,是信息交換編解碼器。在本文的開頭,我就寫到,dubbo將一條消息分成了協(xié)議頭和協(xié)議體,用來解決粘包拆包問題,但是頭跟體在編解碼上有區(qū)別,我們先來看看dubbo 的協(xié)議頭的配置:

上圖是官方文檔的圖片,能夠清晰的看出協(xié)議中各個數(shù)據(jù)所占的位數(shù):

0-7位和8-15位:Magic High和Magic Low,類似java字節(jié)碼文件里的魔數(shù),用來判斷是不是dubbo協(xié)議的數(shù)據(jù)包,就是一個固定的數(shù)字

16位:Req/Res:請求還是響應(yīng)標(biāo)識。

17位:2way:單向還是雙向

18位:Event:是否是事件

19-23位:Serialization 編號

24-31位:status狀態(tài)

32-95位:id編號

96-127位:body數(shù)據(jù)

128-…位:上圖表格內(nèi)的數(shù)據(jù)

可以看到一個該協(xié)議中前65位是協(xié)議頭,后面的都是協(xié)議體數(shù)據(jù)。那么在編解碼中,協(xié)議頭是通過 Codec 編解碼,而body部分是用Serialization序列化和反序列化的。下面我們就來看看該類對協(xié)議頭的編解碼。

1.屬性
// header length.
/**
 * 協(xié)議頭長度:16字節(jié) = 128Bits
 */
protected static final int HEADER_LENGTH = 16;
// magic header.
/**
 * MAGIC二進(jìn)制:1101101010111011,十進(jìn)制:55995
 */
protected static final short MAGIC = (short) 0xdabb;
/**
 * Magic High,也就是0-7位:11011010
 */
protected static final byte MAGIC_HIGH = Bytes.short2bytes(MAGIC)[0];
/**
 * Magic Low  8-15位 :10111011
 */
protected static final byte MAGIC_LOW = Bytes.short2bytes(MAGIC)[1];
// message flag.
/**
 * 128 二進(jìn)制:10000000
 */
protected static final byte FLAG_REQUEST = (byte) 0x80;
/**
 * 64 二進(jìn)制:1000000
 */
protected static final byte FLAG_TWOWAY = (byte) 0x40;
/**
 * 32 二進(jìn)制:100000
 */
protected static final byte FLAG_EVENT = (byte) 0x20;
/**
 * 31 二進(jìn)制:11111
 */
protected static final int SERIALIZATION_MASK = 0x1f;

可以看到 MAGIC是個固定的值,用來判斷是不是dubbo協(xié)議的數(shù)據(jù)包,并且MAGIC_LOW和MAGIC_HIGH分別是MAGIC的低位和高位。其他的屬性用來干嘛后面會講到。

2.encode
@Override
public void encode(Channel channel, ChannelBuffer buffer, Object msg) throws IOException {
    if (msg instanceof Request) {
        // 如果消息是Request類型,對請求消息編碼
        encodeRequest(channel, buffer, (Request) msg);
    } else if (msg instanceof Response) {
        // 如果消息是Response類型,對響應(yīng)消息編碼
        encodeResponse(channel, buffer, (Response) msg);
    } else {
        // 直接讓父類( Telnet ) 處理,目前是 Telnet 命令的結(jié)果。
        super.encode(channel, buffer, msg);
    }
}

該方法是根據(jù)消息的類型來分別進(jìn)行編碼,分為三種情況:Request類型、Response類型以及其他

3.encodeRequest
protected void encodeRequest(Channel channel, ChannelBuffer buffer, Request req) throws IOException {
    Serialization serialization = getSerialization(channel);
    // header.
    // 創(chuàng)建16字節(jié)的字節(jié)數(shù)組
    byte[] header = new byte[HEADER_LENGTH];
    // set magic number.
    // 設(shè)置前16位數(shù)據(jù),也就是設(shè)置header[0]和header[1]的數(shù)據(jù)為Magic High和Magic Low
    Bytes.short2bytes(MAGIC, header);

    // set request and serialization flag.
    // 16-23位為serialization編號,用到或運(yùn)算10000000|serialization編號,例如serialization編號為11111,則為00011111
    header[2] = (byte) (FLAG_REQUEST | serialization.getContentTypeId());

    // 繼續(xù)上面的例子,00011111|1000000 = 01011111
    if (req.isTwoWay()) header[2] |= FLAG_TWOWAY;
    // 繼續(xù)上面的例子,01011111|100000 = 011 11111 可以看到011代表請求標(biāo)記、雙向、是事件,這樣就設(shè)置了16、17、18位,后面19-23位是Serialization 編號
    if (req.isEvent()) header[2] |= FLAG_EVENT;

    // set request id.
    // 設(shè)置32-95位請求id
    Bytes.long2bytes(req.getId(), header, 4);

    // encode request data.
    // // 編碼 `Request.data` 到 Body ,并寫入到 Buffer
    int savedWriteIndex = buffer.writerIndex();
    buffer.writerIndex(savedWriteIndex + HEADER_LENGTH);
    ChannelBufferOutputStream bos = new ChannelBufferOutputStream(buffer);
    // 對body數(shù)據(jù)序列化
    ObjectOutput out = serialization.serialize(channel.getUrl(), bos);
    // 如果該請求是事件
    if (req.isEvent()) {
        // 特殊事件編碼
        encodeEventData(channel, out, req.getData());
    } else {
        // 正常請求編碼
        encodeRequestData(channel, out, req.getData(), req.getVersion());
    }
    // 釋放資源
    out.flushBuffer();
    if (out instanceof Cleanable) {
        ((Cleanable) out).cleanup();
    }
    bos.flush();
    bos.close();
    int len = bos.writtenBytes();
    //檢驗消息長度
    checkPayload(channel, len);
    // 設(shè)置96-127位:Body值
    Bytes.int2bytes(len, header, 12);

    // write
    // 把header寫入到buffer
    buffer.writerIndex(savedWriteIndex);
    buffer.writeBytes(header); // write header.
    buffer.writerIndex(savedWriteIndex + HEADER_LENGTH + len);
}

該方法是對Request類型的消息進(jìn)行編碼,仔細(xì)閱讀上述我寫的注解,結(jié)合協(xié)議頭各個位數(shù)的含義,好好品味我舉的例子。享受二進(jìn)制位運(yùn)算帶來的快樂,也可以看到前半部分邏輯是對協(xié)議頭的編碼,后面還有對body值的序列化。

4.encodeResponse
protected void encodeRequest(Channel channel, ChannelBuffer buffer, Request req) throws IOException {
    Serialization serialization = getSerialization(channel);
    // header.
    // 創(chuàng)建16字節(jié)的字節(jié)數(shù)組
    byte[] header = new byte[HEADER_LENGTH];
    // set magic number.
    // 設(shè)置前16位數(shù)據(jù),也就是設(shè)置header[0]和header[1]的數(shù)據(jù)為Magic High和Magic Low
    Bytes.short2bytes(MAGIC, header);

    // set request and serialization flag.
    // 16-23位為serialization編號,用到或運(yùn)算10000000|serialization編號,例如serialization編號為11111,則為00011111
    header[2] = (byte) (FLAG_REQUEST | serialization.getContentTypeId());

    // 繼續(xù)上面的例子,00011111|1000000 = 01011111
    if (req.isTwoWay()) header[2] |= FLAG_TWOWAY;
    // 繼續(xù)上面的例子,01011111|100000 = 011 11111 可以看到011代表請求標(biāo)記、雙向、是事件,這樣就設(shè)置了16、17、18位,后面19-23位是Serialization 編號
    if (req.isEvent()) header[2] |= FLAG_EVENT;

    // set request id.
    // 設(shè)置32-95位請求id
    Bytes.long2bytes(req.getId(), header, 4);

    // encode request data.
    // // 編碼 `Request.data` 到 Body ,并寫入到 Buffer
    int savedWriteIndex = buffer.writerIndex();
    buffer.writerIndex(savedWriteIndex + HEADER_LENGTH);
    ChannelBufferOutputStream bos = new ChannelBufferOutputStream(buffer);
    // 對body數(shù)據(jù)序列化
    ObjectOutput out = serialization.serialize(channel.getUrl(), bos);
    // 如果該請求是事件
    if (req.isEvent()) {
        // 特殊事件編碼
        encodeEventData(channel, out, req.getData());
    } else {
        // 正常請求編碼
        encodeRequestData(channel, out, req.getData(), req.getVersion());
    }
    // 釋放資源
    out.flushBuffer();
    if (out instanceof Cleanable) {
        ((Cleanable) out).cleanup();
    }
    bos.flush();
    bos.close();
    int len = bos.writtenBytes();
    //檢驗消息長度
    checkPayload(channel, len);
    // 設(shè)置96-127位:Body值
    Bytes.int2bytes(len, header, 12);

    // write
    // 把header寫入到buffer
    buffer.writerIndex(savedWriteIndex);
    buffer.writeBytes(header); // write header.
    buffer.writerIndex(savedWriteIndex + HEADER_LENGTH + len);
}

protected void encodeResponse(Channel channel, ChannelBuffer buffer, Response res) throws IOException {
    int savedWriteIndex = buffer.writerIndex();
    try {
        Serialization serialization = getSerialization(channel);
        // header.
        // 創(chuàng)建16字節(jié)大小的字節(jié)數(shù)組
        byte[] header = new byte[HEADER_LENGTH];
        // set magic number.
        // 設(shè)置前0-15位為魔數(shù)
        Bytes.short2bytes(MAGIC, header);
        // set request and serialization flag.
        // 設(shè)置響應(yīng)標(biāo)志和序列化id
        header[2] = serialization.getContentTypeId();
        // 如果是心跳事件,則設(shè)置第18位為事件
        if (res.isHeartbeat()) header[2] |= FLAG_EVENT;
        // set response status.
        // 設(shè)置24-31位為狀態(tài)碼
        byte status = res.getStatus();
        header[3] = status;
        // set request id.
        // 設(shè)置32-95位為請求id
        Bytes.long2bytes(res.getId(), header, 4);

        // 寫入數(shù)據(jù)
        buffer.writerIndex(savedWriteIndex + HEADER_LENGTH);
        ChannelBufferOutputStream bos = new ChannelBufferOutputStream(buffer);
        // 對body進(jìn)行序列化
        ObjectOutput out = serialization.serialize(channel.getUrl(), bos);
        // encode response data or error message.
        if (status == Response.OK) {
            if (res.isHeartbeat()) {
                // 對心跳事件編碼
                encodeHeartbeatData(channel, out, res.getResult());
            } else {
                // 對普通響應(yīng)編碼
                encodeResponseData(channel, out, res.getResult(), res.getVersion());
            }
        } else out.writeUTF(res.getErrorMessage());
        // 釋放
        out.flushBuffer();
        if (out instanceof Cleanable) {
            ((Cleanable) out).cleanup();
        }
        bos.flush();
        bos.close();

        int len = bos.writtenBytes();
        checkPayload(channel, len);
        Bytes.int2bytes(len, header, 12);
        // write
        buffer.writerIndex(savedWriteIndex);
        buffer.writeBytes(header); // write header.
        buffer.writerIndex(savedWriteIndex + HEADER_LENGTH + len);
    } catch (Throwable t) {
        // clear buffer
        buffer.writerIndex(savedWriteIndex);
        // send error message to Consumer, otherwise, Consumer will wait till timeout.
        //如果在寫入數(shù)據(jù)失敗,則返回響應(yīng)格式錯誤的返回碼
        if (!res.isEvent() && res.getStatus() != Response.BAD_RESPONSE) {
            Response r = new Response(res.getId(), res.getVersion());
            r.setStatus(Response.BAD_RESPONSE);

            if (t instanceof ExceedPayloadLimitException) {
                logger.warn(t.getMessage(), t);
                try {
                    r.setErrorMessage(t.getMessage());
                    // 發(fā)送響應(yīng)
                    channel.send(r);
                    return;
                } catch (RemotingException e) {
                    logger.warn("Failed to send bad_response info back: " + t.getMessage() + ", cause: " + e.getMessage(), e);
                }
            } else {
                // FIXME log error message in Codec and handle in caught() of IoHanndler?
                logger.warn("Fail to encode response: " + res + ", send bad_response info instead, cause: " + t.getMessage(), t);
                try {
                    r.setErrorMessage("Failed to send response: " + res + ", cause: " + StringUtils.toString(t));
                    channel.send(r);
                    return;
                } catch (RemotingException e) {
                    logger.warn("Failed to send bad_response info back: " + res + ", cause: " + e.getMessage(), e);
                }
            }
        }

        // Rethrow exception
        if (t instanceof IOException) {
            throw (IOException) t;
        } else if (t instanceof RuntimeException) {
            throw (RuntimeException) t;
        } else if (t instanceof Error) {
            throw (Error) t;
        } else {
            throw new RuntimeException(t.getMessage(), t);
        }
    }
}

該方法是對Response類型的消息進(jìn)行編碼,該方法里面我沒有舉例子演示如何進(jìn)行編碼,不過過程跟encodeRequest類似。

5.decode
@Override
public Object decode(Channel channel, ChannelBuffer buffer) throws IOException {
    int readable = buffer.readableBytes();
    // 讀取前16字節(jié)的協(xié)議頭數(shù)據(jù),如果數(shù)據(jù)不滿16字節(jié),則讀取全部
    byte[] header = new byte[Math.min(readable, HEADER_LENGTH)];
    buffer.readBytes(header);
    // 解碼
    return decode(channel, buffer, readable, header);
}

@Override
protected Object decode(Channel channel, ChannelBuffer buffer, int readable, byte[] header) throws IOException {
    // check magic number.
    // 核對魔數(shù)(該數(shù)字固定)
    if (readable > 0 && header[0] != MAGIC_HIGH
            || readable > 1 && header[1] != MAGIC_LOW) {
        int length = header.length;
        // 將 buffer 完全復(fù)制到 `header` 數(shù)組中
        if (header.length < readable) {
            header = Bytes.copyOf(header, readable);
            buffer.readBytes(header, length, readable - length)           
               
                                           
                       
                 

文章版權(quán)歸作者所有,未經(jīng)允許請勿轉(zhuǎn)載,若此文章存在違規(guī)行為,您可以聯(lián)系管理員刪除。

轉(zhuǎn)載請注明本文地址:http://specialneedsforspecialkids.com/yun/72797.html

相關(guān)文章

  • dubbo源碼解析(四六)消費(fèi)端發(fā)送請求過程

    摘要:可以參考源碼解析二十四遠(yuǎn)程調(diào)用協(xié)議的八。十六的該類也是用了適配器模式,該類主要的作用就是增加了心跳功能,可以參考源碼解析十遠(yuǎn)程通信層的四。二十的可以參考源碼解析十七遠(yuǎn)程通信的一。 2.7大揭秘——消費(fèi)端發(fā)送請求過程 目標(biāo):從源碼的角度分析一個服務(wù)方法調(diào)用經(jīng)歷怎么樣的磨難以后到達(dá)服務(wù)端。 前言 前一篇文章講到的是引用服務(wù)的過程,引用服務(wù)無非就是創(chuàng)建出一個代理。供消費(fèi)者調(diào)用服務(wù)的相關(guān)方法。...

    fish 評論0 收藏0
  • dubbo源碼解析(四七)服務(wù)端處理請求過程

    摘要:而存在的意義就是保證請求或響應(yīng)對象可在線程池中被解碼,解碼完成后,就會分發(fā)到的。 2.7大揭秘——服務(wù)端處理請求過程 目標(biāo):從源碼的角度分析服務(wù)端接收到請求后的一系列操作,最終把客戶端需要的值返回。 前言 上一篇講到了消費(fèi)端發(fā)送請求的過程,該篇就要將服務(wù)端處理請求的過程。也就是當(dāng)服務(wù)端收到請求數(shù)據(jù)包后的一系列處理以及如何返回最終結(jié)果。我們也知道消費(fèi)端在發(fā)送請求的時候已經(jīng)做了編碼,所以我...

    yzzz 評論0 收藏0
  • dubbo源碼解析(八)遠(yuǎn)程通信——開篇

    摘要:而編碼器是講應(yīng)用程序的數(shù)據(jù)轉(zhuǎn)化為網(wǎng)絡(luò)格式,解碼器則是講網(wǎng)絡(luò)格式轉(zhuǎn)化為應(yīng)用程序,同時具備這兩種功能的單一組件就叫編解碼器。在中是老的編解碼器接口,而是新的編解碼器接口,并且已經(jīng)用把適配成了。 遠(yuǎn)程通訊——開篇 目標(biāo):介紹之后解讀遠(yuǎn)程通訊模塊的內(nèi)容如何編排、介紹dubbo-remoting-api中的包結(jié)構(gòu)設(shè)計以及最外層的的源碼解析。 前言 服務(wù)治理框架中可以大致分為服務(wù)通信和服務(wù)管理兩個...

    Faremax 評論0 收藏0
  • dubbo源碼解析(四八)異步化改造

    摘要:大揭秘異步化改造目標(biāo)從源碼的角度分析的新特性中對于異步化的改造原理??丛创a解析四十六消費(fèi)端發(fā)送請求過程講到的十四的,在以前的邏輯會直接在方法中根據(jù)配置區(qū)分同步異步單向調(diào)用。改為關(guān)于可以參考源碼解析十遠(yuǎn)程通信層的六。 2.7大揭秘——異步化改造 目標(biāo):從源碼的角度分析2.7的新特性中對于異步化的改造原理。 前言 dubbo中提供了很多類型的協(xié)議,關(guān)于協(xié)議的系列可以查看下面的文章: du...

    lijinke666 評論0 收藏0
  • dubbo源碼解析(四四)服務(wù)暴露過程

    摘要:服務(wù)暴露過程目標(biāo)從源碼的角度分析服務(wù)暴露過程。導(dǎo)出服務(wù),包含暴露服務(wù)到本地,和暴露服務(wù)到遠(yuǎn)程兩個過程。其中服務(wù)暴露的第八步已經(jīng)沒有了。將泛化調(diào)用版本號或者等信息加入獲得服務(wù)暴露地址和端口號,利用內(nèi)數(shù)據(jù)組裝成。 dubbo服務(wù)暴露過程 目標(biāo):從源碼的角度分析服務(wù)暴露過程。 前言 本來這一篇一個寫異步化改造的內(nèi)容,但是最近我一直在想,某一部分的優(yōu)化改造該怎么去撰寫才能更加的讓讀者理解。我覺...

    light 評論0 收藏0

發(fā)表評論

0條評論

最新活動
閱讀需要支付1元查看
<