Tars-Java客戶端源碼分析

時間 2021-05-08

標籤 java git github 算法編程 api 緩存服務器網絡 session 欄目 Java 简体版

原文原文鏈接

1、基本RPC框架簡介

在分佈式計算中，遠程過程調用（Remote Procedure Call，縮寫 RPC）容許運行於一臺計算機的程序調用另外一個地址空間計算機的程序，就像調用本地程序同樣，無需額外地爲這個交互做用涉及到的代理對象構建、網絡協議等進行編程。java

通常RPC架構，有至少三種結構，分別爲註冊中心，服務提供者和服務消費者。如圖1.1所示，註冊中心提供註冊服務和註冊信息變動的通知服務，服務提供者運行在服務器來提供服務，服務消費者使用服務提供者的服務。git

服務提供者（RPC Server），運行在服務端，提供服務接口定義與服務實現類，並對外暴露服務接口。註冊中心（Registry），運行在服務端，負責記錄服務提供者的服務對象，並提供遠程服務信息的查詢服務和變動通知服務。服務消費者（RPC Client），運行在客戶端，經過遠程代理對象調用遠程服務。github

1.1 RPC調用流程

以下圖所示，描述了RPC的調用流程，其中IDL(Interface Description Language)爲接口描述語言，使得在不一樣平臺上運行的程序和用不一樣語言編寫的程序能夠相互通訊交流。算法

1）客戶端調用客戶端樁模塊。該調用是本地過程調用，其中參數以正常方式推入堆棧。編程

2）客戶端樁模塊將參數打包到消息中，並進行系統調用以發送消息。打包參數稱爲編組。api

3）客戶端的本地操做系統將消息從客戶端計算機發送到服務器計算機。緩存

4）服務器計算機上的本地操做系統將傳入的數據包傳遞到服務器樁模塊。服務器

5）服務器樁模塊從消息中解包出參數。解包參數稱爲解組。網絡

6）最後，服務器樁模塊執行服務器程序流程。回覆是沿相反的方向執行相同的步驟。session

2、Tars Java客戶端設計介紹

Tars Java客戶端總體設計與主流的RPC框架基本一致。咱們先介紹Tars Java客戶端初始化過程。

2.1 Tars Java客戶端初始化過程

如圖2.1所示，描述了Tars Java的初始化過程。

1）先出建立一個CommunicatorConfig配置項，命名爲communicatorConfig，其中按需設置locator, moduleName, connections等參數。

2）經過上述的CommunicatorConfig配置項，命名爲config，那麼調用CommunicatorFactory.getInstance().getCommunicator(config)，建立一個Communicator對象，命名爲communicator。

3）假設objectName="MESSAGE.ControlCenter.Dispatcher"，須要生成的代理接口爲Dispatcher.class，調用communicator.stringToProxy(objectName, Dispatcher.class)方法來生成代理對象的實現類。

4）在stringToProxy()方法裏，首先經過初始化QueryHelper代理對象，調用getServerNodes()方法獲取遠程服務對象列表，並設置該返回值到communicatorConfig的objectName字段裏。具體的代理對象的代碼分析，見下文中的「2.3 代理生成」章節。

5）判斷在以前調用stringToProxy是否有設置LoadBalance參數，若是沒有的話，就生成默認的採用RR輪訓算法的DefaultLoadBalance對象。

6）建立TarsProtocolInvoker協議調用對象，其中過程有經過解析communicatorConfig中的objectName和simpleObjectName來獲取URL列表，其中一個URL對應一個遠程服務對象，TarsProtocolInvoker初始化各個URL對應的ServantClient對象，其中一個URL根據communicatorConfig的connections配置項確認生成多少個ServantClient對象。而後使用ServantClients等參數初始化TarsInvoker對象，並將這些TarsInvoker對象集合設置到TarsProtocolInvoker的allInvokers成員變量中，其中每一個URL對應一個TarsInvoker對象。上述分析代表，一個遠程服務節點對應一個TarsInvoker對象，一個TarsInvoker對象包含connections個ServantClient對象，對於TCP協議，那麼就是一個ServantClient對象對應一個TCP鏈接。

7）使用api, objName, servantProxyConfig,loadBalance,protocolInvoker, this.communicator參數生成一個實現JDK代理接口InvocationHandler的ObjectProxy對象。

8）生成ObjectProxy對象的同時進行初始化操做，首先會執行loadBalancer.refresh()方法刷新遠程服務節點到負載均衡器中便於後續tars遠程調用進行路由。

9）而後註冊統計信息上報器，其中是上報方法採用JDK的ScheduledThreadPoolExecutor進行定時輪訓上報。

10）註冊服務列表刷新器，採用的技術方法和上述統計信息上報器基本一致。

2.2 使用範例

如下代碼爲最簡化示例，其中CommunicatorConfig裏的配置採用默認值，communicator經過CommunicatorConfig配置生成後，直接指定遠程服務對象的具體服務對象名、IP和端口生成一個遠程服務代理對象。

Tars Java代碼使用範例// 先初始化基本Tars配置CommunicatorConfig cfg = new CommunicatorConfig();// 經過上述的CommunicatorConfig配置生成一個Communicator對象。Communicator communicator = CommunicatorFactory.getInstance().getCommunicator(cfg);// 指定Tars遠程服務的服務對象名、IP和端口生成一個遠程服務代理對象。

// 先初始化基本Tars配置
    CommunicatorConfig cfg = new CommunicatorConfig();
    // 經過上述的CommunicatorConfig配置生成一個Communicator對象。
    Communicator communicator = CommunicatorFactory.getInstance().getCommunicator(cfg);
    // 指定Tars遠程服務的服務對象名、IP和端口生成一個遠程服務代理對象。
    HelloPrx proxy = communicator.stringToProxy(HelloPrx.class, "TestApp.HelloServer.HelloObj@tcp -h 127.0.0.1 -p 18601 -t 60000");
    //同步調用，阻塞直到遠程服務對象的方法返回結果
    String ret = proxy.hello(3000, "Hello World");
    System.out.println(ret);
    //異步調用，不關注異步調用最終的狀況
    proxy.async_hello(null, 3000, "Hello World");
      //異步調用,註冊一個實現TarsAbstractCallback接口的回執處理對象，該實現類分別處理調用成功，調用超時和調用異常的狀況。
    proxy.async_hello(new HelloPrxCallback() {
        @Override
        public void callback_expired() { //超時事件處理
        }
        @Override
        public void callback_exception(Throwable ex) { //異常事件處理
        }
        @Override
        public void callback_hello(String ret) { //調用成功事件處理
            Main.logger.info("invoke async method successfully {}", ret);
       }
    }, 1000, "Hello World");

在上述例子中，演示了常見的兩種調用方式，分別爲同步調用和異步調用。其中異步調用，若是調用方想捕捉異步調用的最終結果，能夠註冊一個實現TarsAbstractCallback接口的實現類，對tars調用的異常，超時和成功事件進行處理。

2.3 代理生成

Tars Java的客戶端樁模塊的遠程代理對象是採用JDK原生Proxy方法。以下文的源碼所示，ObjectProxy實現了java.lang.reflect.InvocationHandler的接口方法，該接口是JDK自帶的代理接口。

代理實現

public final class ObjectProxy<T> implements ServantProxy, InvocationHandler {
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        String methodName = method.getName();
        Class<?>[] parameterTypes = method.getParameterTypes();
        InvokeContext context = this.protocolInvoker.createContext(proxy, method, args);
        try {
            if ("toString".equals(methodName) && parameterTypes.length == 0) {
                return this.toString();
            } else if
                //***** 省略代碼 *****
            } else {
                // 在負載均衡器選取一個遠程調用類，進行應用層協議的封裝，最後調用TCP傳輸層進行發送。
                Invoker invoker = this.loadBalancer.select(context);
                return invoker.invoke(context);
            }
        } catch (Throwable var8) {
            // ***** 省略代碼 *****
        }
    }
}

固然生成上述遠程服務代理類，涉及到輔助類，Tars Java採用ServantProxyFactory來生成上述的ObjectProxy，並存儲ObjectProxy對象到Map結構，便於調用方二次使用時直接複用已存在的遠程服務代理對象。

具體相關邏輯如源碼所示，ObjectProxyFactory是生成ObjectProxy的輔助工廠類，和ServantProxyFactory不一樣，其自己不緩存生成的代理對象。

class ServantProxyFactory {
    private final ConcurrentHashMap<String, Object> cache = new ConcurrentHashMap();
    // ***** 省略代碼 *****
    public <T> Object getServantProxy(Class<T> clazz, String objName, ServantProxyConfig servantProxyConfig, LoadBalance loadBalance, ProtocolInvoker<T> protocolInvoker) {
        Object proxy = this.cache.get(objName);
        if (proxy == null) {
            this.lock.lock(); // 加鎖，保證只生成一個遠程服務代理對象。
            try {
                proxy = this.cache.get(objName);
                if (proxy == null) {
                    // 建立實現JDK的java.lang.reflect.InvocationHandler接口的對象
                    ObjectProxy<T> objectProxy = this.communicator.getObjectProxyFactory().getObjectProxy(clazz, objName, servantProxyConfig, loadBalance, protocolInvoker);
                    // 使用JDK的java.lang.reflect.Proxy來生成實際的代理對象
                    this.cache.putIfAbsent(objName, this.createProxy(clazz, objectProxy));
                    proxy = this.cache.get(objName);
                }
            } finally {
                this.lock.unlock();
            }
        }
        return proxy;
    }
    /** 使用JDK自帶的Proxy.newProxyInstance生成代理對象 */
    private <T> Object createProxy(Class<T> clazz, ObjectProxy<T> objectProxy) {
        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), new Class[]{clazz, ServantProxy.class}, objectProxy);
    }
    // ***** 省略代碼 *****
}

從以上的源碼中，能夠看到createProxy使用了JDK的Proxy.newProxyInstance方法來生成遠程服務代理對象。

2.4 遠程服務尋址方法

做爲一個RPC遠程框架，在分佈式系統中，調用遠程服務，涉及到如何路由的問題，也就是如何從多個遠程服務節點中選擇一個服務節點進行調用，固然Tars Java支持直連特定節點的方式調用遠程服務，如上文的2.2 使用範例所介紹。

如圖下圖所示，ClientA某個時刻的一次調用使用了Service3節點進行遠程服務調用，而ClientB某個時刻的一次調用採用Service2節點。Tars Java提供多種負載均衡算法實現類，其中有采用RR輪訓算法的RoundRobinLoadBalance，一致性哈希算法的ConsistentHashLoadBalance和普通哈希算法的HashLoadBalance。

以下述源碼所示，若是要自定義負載均衡器來定義遠程調用的路由規則，那麼須要實現com.qq.tars.rpc.common.LoadBalance接口，其中LoadBalance.select()方法負責按照路由規則，選取對應的Invoker對象，而後進行遠程調用，具體邏輯見源碼代理實現。因爲遠程服務節點可能發生變動，好比上下線遠程服務節點，須要刷新本地負載均衡器的路由信息，那麼此信息更新的邏輯在LoadBalance.refresh()方法裏實現。

負載均衡接口

public interface LoadBalance<T> {
    /** 根據負載均衡策略，挑選invoker */
    Invoker<T> select(InvokeContext invokeContext) throws NoInvokerException;
    /** 通知invoker列表的更新 */
    void refresh(Collection<Invoker<T>> invokers);
}

2.5 網絡模型

Tars Java的IO模式採用的JDK的NIO的Selector模式。這裏以TCP協議來描述網絡處理，以下述源碼所示，Reactor是一個線程，其中的run()方法中，調用了selector.select()方法，意思是若是除非此時網絡產生一個事件，不然將一直線程阻塞下去。

假如此時出現一個網絡事件，那麼此時線程將會被喚醒，執行後續代碼，其中一個代碼是dispatcheEvent(key)，也就是將進行事件的分發。

其中將根據對應條件，調用acceptor.handleConnectEvent(key)方法來處理客戶端鏈接成功事件，或acceptor.handleAcceptEvent(key)方法來處理服務器接受鏈接成功事件，或調用acceptor.handleReadEvent(key)方法從Socket裏讀取數據，或acceptor.handleWriteEvent(key)方法來寫數據到Socket 。

Reactor事件處理

public final class Reactor extends Thread {
    protected volatile Selector selector = null;
    private Acceptor acceptor = null;
    //***** 省略代碼 *****
    public void run() {
        try {
            while (!Thread.interrupted()) {
                // 阻塞直到有網絡事件發生。
                selector.select();
                //***** 省略代碼 *****
                while (iter.hasNext()) {
                    SelectionKey key = iter.next();
                    iter.remove();
                    if (!key.isValid()) continue;
                    try {
                        //***** 省略代碼 *****
                        // 分發傳輸層協議TCP或UDP網絡事件
                        dispatchEvent(key);
                //***** 省略代碼 *****
            }
        }
        //***** 省略代碼 *****
    }
        //***** 省略代碼 *****
    private void dispatchEvent(final SelectionKey key) throws IOException {
        if (key.isConnectable()) {
            acceptor.handleConnectEvent(key);
        } else if (key.isAcceptable()) {
            acceptor.handleAcceptEvent(key);
        } else if (key.isReadable()) {
            acceptor.handleReadEvent(key);
        } else if (key.isValid() && key.isWritable()) {
            acceptor.handleWriteEvent(key);
        }
    }
}

網絡處理採用Reactor事件驅動模式，Tars定義一個Reactor對象對應一個Selector對象，針對每一個遠程服務（總體服務集羣，非單個節點程序）默認建立2個Reactor對象進行處理，經過修改com.qq.tars.net.client.selectorPoolSize這個JVM啓動參數值來決定一個遠程服務具體建立幾個Reactor對象。

上圖中的處理讀IO事件（Read Event）實現和寫IO事件（Write Event）的線程池是在Communicator初始化的時候配置的。具體邏輯如源碼所示，其中線程池參數配置由CommunicatorConfig的corePoolSize, maxPoolSize, keepAliveTime等參數決定。

讀寫事件線程池初始化

private void initCommunicator(CommunicatorConfig config) throws CommunicatorConfigException {
    //***** 省略代碼 *****
    this.threadPoolExecutor = ClientPoolManager.getClientThreadPoolExecutor(config);
    //***** 省略代碼 *****
}

public class ClientPoolManager {
    public static ThreadPoolExecutor getClientThreadPoolExecutor(CommunicatorConfig communicatorConfig) {
        //***** 省略代碼 *****
        clientThreadPoolMap.put(communicatorConfig, createThreadPool(communicatorConfig));
        //***** 省略代碼 *****
        return clientPoolExecutor;
    }    
     
    private static ThreadPoolExecutor createThreadPool(CommunicatorConfig communicatorConfig) {
        int corePoolSize = communicatorConfig.getCorePoolSize();
        int maxPoolSize = communicatorConfig.getMaxPoolSize();
        int keepAliveTime = communicatorConfig.getKeepAliveTime();
        int queueSize = communicatorConfig.getQueueSize();
        TaskQueue taskqueue = new TaskQueue(queueSize);

        String namePrefix = "tars-client-executor-";
        TaskThreadPoolExecutor executor = new TaskThreadPoolExecutor(corePoolSize, maxPoolSize, keepAliveTime, TimeUnit.SECONDS, taskqueue, new TaskThreadFactory(namePrefix));
        taskqueue.setParent(executor);
        return executor;
    }
}

2.6 遠程調用交互模型

調用代理類的方法，那麼會進入實現InvocationHandler接口的ObjectProxy中的invoke方法。

下圖描述了遠程服務調用的流程狀況。這裏着重講幾個點，一個是如何寫數據到網絡IO。第二個是Tars Java經過什麼方式進行同步或者異步調用，底層採用了什麼技術。

2.6.1 寫 IO 流程

如圖（底層代碼寫IO過程）所示，ServantClient將調用底層網絡寫操做，在invokeWithSync方法中，取得ServantClient自身成員變量TCPSession，調用TCPSession.write()方法，如圖（底層代碼寫IO過程）和如下源碼（讀寫事件線程池初始化）所示，先獲取Encode進行請求內容編碼成IoBuffer對象，最後將IoBuffer的java.nio.ByteBuffer內容放入TCPSession的queue成員變量中，而後調用key.selector().wakeup()，喚醒Reactor中run()方法中的Selector.select()，執行後續的寫操做。

具體Reactor邏輯見上文2.5 網絡模型內容，若是Reactor檢查條件發現能夠寫IO的話也就是key.isWritable()爲true，那麼最終會循環從TCPSession.queue中取出ByteBuffer對象，調用SocketChannel.write(byteBuffer)執行實際的寫網絡Socket操做，代碼邏輯見源碼中的doWrite()方法。

讀寫事件線程池初始化

public class TCPSession extends Session {
    public void write(Request request) throws IOException {
        try {
            IoBuffer buffer = selectorManager.getProtocolFactory().getEncoder().encodeRequest(request, this);
            write(buffer);
        //***** 省略代碼 *****
    }
    protected void write(IoBuffer buffer) throws IOException {
        //***** 省略代碼 *****
        if (!this.queue.offer(buffer.buf())) {
            throw new IOException("The session queue is full. [ queue size:" + queue.size() + " ]");
        }
        if (key != null) {
            key.interestOps(key.interestOps() | SelectionKey.OP_WRITE);
            key.selector().wakeup();
        }
    }
    protected synchronized int doWrite() throws IOException {
        int writeBytes = 0;
        while (true) {
            ByteBuffer wBuf = queue.peek();
            //***** 省略代碼 *****
            int bytesWritten = ((SocketChannel) channel).write(wBuf);
            //***** 省略代碼 *****
        return writeBytes;
    }
}

2.6.2 同步和異步調用的底層技術實現

對於同步方法調用，如圖（遠程調用流程）和源碼（ServantClient的同步調用）所示，ServantClient調用底層網絡寫操做，在invokeWithSync方法中建立一個Ticket對象，Ticket顧名思義就是票的意思，這張票惟一標識本次網絡調用狀況。

ServantClient的同步調用

public class ServantClient {
    public <T extends ServantResponse> T invokeWithSync(ServantRequest request) throws IOException {
            //***** 省略代碼 *****
            ticket = TicketManager.createTicket(request, session, this.syncTimeout);
            Session current = session;
            current.write(request);
            if (!ticket.await(this.syncTimeout, TimeUnit.MILLISECONDS)) {
            //***** 省略代碼 *****
            response = ticket.response();
            //***** 省略代碼 *****
            return response;
            //***** 省略代碼 *****
        return response;
    }
}

如代碼所示，在執行完session.write()操做後，緊接着執行ticket.await()方法，該方法線程等待直到遠程服務回覆返回結果到客戶端，ticket.await()被喚醒後，將執行後續操做，最終invokeWithSync方法返回response對象。其中Ticket的等待喚醒功能內部採用java.util.concurrent.CountDownLatch來實現。

對於異步方法調用，將會執行ServantClient.invokeWithAsync方法，也會建立一個Ticket，而且執行Session.write()操做，雖然不會調用ticket.await()，可是在Reactor接收到遠程回覆時，首先會先解析Tars協議頭獲得Response對象，而後將Response對象放入如圖（Tars-Java的網絡事件處理模型）所示的IO讀寫線程池中進行進一步處理，以下述源碼（異步回調事件處理）所示，最終會調用WorkThread.run()方法，在run()方法裏執行ticket.notifyResponse(resp)，該方法裏面會執行相似上述代碼2.1中的實現TarsAbstractCallback接口的調用成功回調的方法。

異步回調事件處理

public final class WorkThread implements Runnable {
    public void run() {
        try {
            //***** 省略代碼 *****
                Ticket<Response> ticket = TicketManager.getTicket(resp.getTicketNumber());
            //***** 省略代碼 *****
                ticket.notifyResponse(resp);
                ticket.countDown();
                TicketManager.removeTicket(ticket.getTicketNumber());
            }
            //***** 省略代碼 *****
    }
}

以下述源碼所示，TicketManager會有一個定時任務輪訓檢查全部的調用是否超時，若是(currentTime - t.startTime) > t.timeout條件成立，那麼會調用t.expired()告知回調對象，本次調用超時。

調用超時事件處理

public class TicketManager {
            //***** 省略代碼 *****
    static {
        executor.scheduleAtFixedRate(new Runnable() {
            long currentTime = -1;
            public void run() {
                Collection<Ticket<?>> values = tickets.values();
                currentTime = System.currentTimeMillis();
                for (Ticket<?> t : values) {
                    if ((currentTime - t.startTime) > t.timeout) {
                        removeTicket(t.getTicketNumber());
                        t.expired();
                    }
                }
            }
        }, 500, 500, TimeUnit.MILLISECONDS);
    }
}