Tars-Java客戶端源碼分析
一、基本RPC框架簡介
在分散式計算中，遠程過程調用（Remote Procedure Call，縮寫 RPC）允許運行於一台電腦的程式調用另一個地址空間電腦的程式，就像調用本地程式一樣，無需額外地為這個交互作用涉及到的代理對象構建、網路協議等進行編程。
一般RPC架構，有至少三種結構，分別為註冊中心，服務提供者和服務消費者。如圖1.1所示，註冊中心提供註冊服務和註冊資訊變更的通知服務，服務提供者運行在伺服器來提供服務，服務消費者使用服務提供者的服務。
服務提供者（RPC Server），運行在服務端，提供服務介面定義與服務實現類，並對外暴露服務介面。註冊中心（Registry），運行在服務端，負責記錄服務提供者的服務對象，並提供遠程服務資訊的查詢服務和變更通知服務。服務消費者（RPC Client），運行在客戶端，通過遠程代理對象調用遠程服務。
1.1 RPC調用流程
如下圖所示，描述了RPC的調用流程，其中IDL(Interface Description Language)為介面描述語言，使得在不同平台上運行的程式和用不同語言編寫的程式可以相互通訊交流。
1）客戶端調用客戶端樁模組。該調用是本地過程調用，其中參數以正常方式推入堆棧。
2）客戶端樁模組將參數打包到消息中，並進行系統調用以發送消息。打包參數稱為編組。
3）客戶端的本地作業系統將消息從客戶端電腦發送到伺服器電腦。
4）伺服器電腦上的本地作業系統將傳入的數據包傳遞到伺服器樁模組。
5）伺服器樁模組從消息中解包出參數。解包參數稱為解組。
6）最後，伺服器樁模組執行伺服器程式流程。回復是沿相反的方向執行相同的步驟。
二、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);
}
}
三、總結
程式碼的調用一般都是層層遞歸調用，程式碼的調用深度和廣度都很大，通過調試程式碼的方式一步步學習源碼的方式，更加容易理解源碼的含義和設計理念。
Tars與其他RPC框架，並沒有什麼本質區別，通過類比其他框架的設計理念，可以更加深入理解Tars Java設計理念。
四、參考文獻
作者：vivo 互聯網伺服器團隊-Ke Shengkai
