一、Tars框架基本介紹

Tars是騰訊開(kāi)源的支持多語(yǔ)言的高性能RPC框架，起源于騰訊內(nèi)部2008年至今一直使用的統(tǒng)一應(yīng)用框架TAF（Total Application Framework），目前支持C++、Java、PHP、Nodejs、Go語(yǔ)言。

(資料圖)

該框架為用戶提供了涉及到開(kāi)發(fā)、運(yùn)維、以及測(cè)試的一整套解決方案，幫助一個(gè)產(chǎn)品或者服務(wù)快速開(kāi)發(fā)、部署、測(cè)試、上線。它集可擴(kuò)展協(xié)議編解碼、高性能RPC通信框架、名字路由與發(fā)現(xiàn)、發(fā)布監(jiān)控、日志統(tǒng)計(jì)、配置管理等于一體，通過(guò)它可以快速用微服務(wù)的方式構(gòu)建自己的穩(wěn)定可靠的分布式應(yīng)用，并實(shí)現(xiàn)完整有效的服務(wù)治理。

官方倉(cāng)庫(kù)地址：

??https://github.com/TarsCloud/Tars??

vivo推送平臺(tái)也深度使用了該框架，部署服務(wù)節(jié)點(diǎn)超過(guò)一千個(gè)，經(jīng)過(guò)線上每日一百多億消息推送量的考驗(yàn)。

此前已在vivo互聯(lián)網(wǎng)技術(shù)公眾號(hào)發(fā)布過(guò)《??Tars Java 客戶端源碼分析??》此篇文章為續(xù)集。

Tars-java 最新穩(wěn)定版1.7.2以及之前的版本都使用Java NIO進(jìn)行網(wǎng)絡(luò)編程；本文將分別詳細(xì)介紹java NIO的原理和Tars 使用NIO進(jìn)行網(wǎng)絡(luò)編程的細(xì)節(jié)。

二、Java NIO原理介紹

從1.4版本開(kāi)始，Java提供了一種新的IO處理方式：NIO (New IO 或 Non-blocking IO)是一個(gè)可以替代標(biāo)準(zhǔn)Java IO 的API，它是面向緩沖區(qū)而不是字節(jié)流，它是非阻塞的，支持IO多路復(fù)用。

2.1 Channels (通道) and Buffers (緩沖區(qū))

標(biāo)準(zhǔn)的IO基于字節(jié)流進(jìn)行操作的，而NIO是基于通道（Channel）和緩沖區(qū)（Buffer）進(jìn)行操作。數(shù)據(jù)總是從通道讀取到緩沖區(qū)中，或者從緩沖區(qū)寫入到通道中，下圖是一個(gè)完整流程。

Channel類型：

支持文件讀寫數(shù)據(jù)的FileChannel能通過(guò)UDP讀寫網(wǎng)絡(luò)中的數(shù)據(jù)的DatagramChannel能通過(guò)TCP讀寫網(wǎng)絡(luò)數(shù)據(jù)的SocketChannel可以監(jiān)聽(tīng)新進(jìn)來(lái)的TCP連接，對(duì)每一個(gè)新進(jìn)來(lái)的連接都會(huì)創(chuàng)建一個(gè)SocketChannel的ServerSocketChannel 。

SocketChannel：

打開(kāi) SocketChannel：SocketChannel socketChannel = SocketChannel.open();關(guān)閉 SocketChannel：socketChannel.close();從Channel中讀取的數(shù)據(jù)放到Buffer：int bytesRead = inChannel.read(buf);將Buffer中的數(shù)據(jù)寫到Channel：int bytesWritten = inChannel.write(buf);

ServerSocketChannel：

通過(guò) ServerSocketChannel.accept()方法監(jiān)聽(tīng)新進(jìn)來(lái)的連接，當(dāng)accept()方法返回的時(shí)候，它返回一個(gè)包含新進(jìn)來(lái)的連接的SocketChannel，因此accept()方法會(huì)一直阻塞到有新連接到達(dá)。

通常不會(huì)僅僅只監(jiān)聽(tīng)一個(gè)連接,在while循環(huán)中調(diào)用 accept()方法. 如下面的例子：

代碼1：

while(true){    SocketChannel socketChannel = serverSocketChannel.accept();     //do something with socketChannel...}

ServerSocketChannel可以設(shè)置成非阻塞模式。在非阻塞模式下，accept() 方法會(huì)立刻返回，如果還沒(méi)有新進(jìn)來(lái)的連接,返回的將是null。因此，需要檢查返回的SocketChannel是否是null。

代碼2：

ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();serverSocketChannel.socket().bind(new InetSocketAddress(8888));serverSocketChannel.configureBlocking(false);while(true){    SocketChannel socketChannel = serverSocketChannel.accept();    if(socketChannel != null){        //do something with socketChannel...    }}

Buffer類型：

ByteBufferCharBufferDoubleBufferFloatBufferIntBufferLongBufferShortBuffer

Buffer的分配：

ByteBuffer buf = ByteBuffer.allocate(2048);

Buffer的讀寫：

一般是以下四個(gè)步驟：

寫入數(shù)據(jù)到Buffer，最大寫入量是capacity，寫模式下limit值即為capacity值，position即為寫到的位置。調(diào)用flip()方法將Buffer從寫模式切換到讀模式，此時(shí)position移動(dòng)到開(kāi)始位置0，limit移動(dòng)到position的位置。從Buffer中讀取數(shù)據(jù)，在讀模式下可以讀取之前寫入到buffer的所有數(shù)據(jù)，即為limit位置。調(diào)用clear()方法或者compact()方法。clear()方法將position設(shè)為0，limit被設(shè)置成capacity的值。compact()方法將所有未讀的數(shù)據(jù)拷貝到Buffer起始處，然后將position設(shè)到最后一個(gè)未讀元素后面。

mark()與reset()方法通過(guò)調(diào)用Buffer.mark()方法，可以標(biāo)記Buffer中的一個(gè)特定position，之后可以通過(guò)調(diào)用Buffer.reset()方法恢復(fù)到這個(gè)position。

duplicate()

此方法返回承載先前字節(jié)緩沖區(qū)內(nèi)容的新字節(jié)緩沖區(qū)。

remaining()limit減去position的值

2.2 Selector（選擇器）

Java NIO引入了選擇器的概念，選擇器用于監(jiān)聽(tīng)多個(gè)通道的事件。單個(gè)的線程可以監(jiān)聽(tīng)多個(gè)數(shù)據(jù)通道。要使用Selector，得向Selector注冊(cè)Channel，然后調(diào)用它的select()方法。這個(gè)方法會(huì)一直阻塞到某個(gè)注冊(cè)的通道有事件就緒。一旦這個(gè)方法返回，線程就可以處理這些事件。

線程使用一個(gè)selector處理多個(gè)channel

代碼3：

channel.configureBlocking(false);SelectionKey key = channel.register(selector,Selectionkey.OP_READ);

注意register()方法的第二個(gè)參數(shù)，這是一個(gè)監(jiān)聽(tīng)的集合，即在通過(guò)Selector監(jiān)聽(tīng)Channel時(shí)關(guān)注什么事件集合。

SelectionKey包含：

1) interest集合：selectionKey.interestOps()可以監(jiān)聽(tīng)四種不同類型的事件：OP_ACCEPT、OP_CONNECT、OP_WRITE、OP_READ

2) ready集合：selectionKey.readyOps(); ready 集合是通道已經(jīng)準(zhǔn)備就緒的操作的集合，提供4個(gè)方便的方法：

selectionKey.isAcceptable();selectionKey.isConnectable();selectionKey.isReadable();selectionKey.isWritable();

3) Channel：selectionKey.channel();

4) Selector：selectionKey.selector();

5)可選的附加對(duì)象：

selectionKey.attachment();可以將一個(gè)對(duì)象或者更多信息附著到SelectionKey上，這樣就能方便的識(shí)別特定的通道。

提示：

OP_ACCEPT和OP_CONNECT的區(qū)別：簡(jiǎn)單來(lái)說(shuō)，客戶端建立連接是connect，服務(wù)器準(zhǔn)備接收連接是accept。一個(gè)典型的客戶端服務(wù)器網(wǎng)絡(luò)交互流程如下圖

selectedKeys()

一旦調(diào)用了select()方法，并且返回值表明有一個(gè)或更多個(gè)通道就緒了，然后可以通過(guò)調(diào)用selector的selectedKeys()方法，訪問(wèn)已選擇鍵集（selected key set）中的就緒通道。

wakeUp()某個(gè)線程調(diào)用select()方法后阻塞了，即使沒(méi)有通道已經(jīng)就緒，也有辦法讓其從select()方法返回。只要讓其它線程在阻塞線程調(diào)用select()方法的對(duì)象上調(diào)用Selector.wakeup()方法即可。阻塞在select()方法上的線程會(huì)立馬返回。如果有其它線程調(diào)用了wakeup()方法，但當(dāng)前沒(méi)有線程阻塞在select()方法上，下個(gè)調(diào)用select()方法的線程會(huì)立即wake up。

close()用完Selector后調(diào)用其close()方法會(huì)關(guān)閉該Selector，且使注冊(cè)到該Selector上的所有SelectionKey實(shí)例無(wú)效。通道本身并不會(huì)關(guān)閉。

通過(guò)Selector選擇通道：

int select()阻塞直到至少有一個(gè)通道在你注冊(cè)的事件上就緒了int select(long timeout)增加最長(zhǎng)阻塞毫秒數(shù)int selectNow()不會(huì)阻塞，不管什么通道就緒都立刻返回三、 Tars NIO網(wǎng)絡(luò)編程

了解完 Java NIO的原理，我們來(lái)看看Tars是如何使用NIO進(jìn)行網(wǎng)絡(luò)編程的。

Tars的網(wǎng)絡(luò)模型是多reactor多線程模型。有一點(diǎn)特殊的是tars的reactor線程組里隨機(jī)選一個(gè)線程處理網(wǎng)絡(luò)事件，并且該線程同時(shí)也能處理讀寫。

核心類之間的關(guān)系如下：

3.1 一個(gè)典型的Java NIO服務(wù)端開(kāi)發(fā)流程創(chuàng)建ServerSocketChannel,設(shè)置為非阻塞，并綁定端口創(chuàng)建Selector對(duì)象給ServerSocketChannel注冊(cè)SelectionKey.OP_ACCEPT事件啟動(dòng)一個(gè)線程循環(huán)，調(diào)用Selector的select方法來(lái)檢查IO就緒事件，一旦有IO就緒事件，就通知用戶線程去處理IO事件如果有Accept事件，就創(chuàng)建一個(gè)SocketChannel，并注冊(cè)SelectionKey.OP_READ如果有讀事件，判斷一下是否全包，如果全包，就交給后端線程處理寫事件比較特殊。isWriteable表示的是本機(jī)的寫緩沖區(qū)是否可寫。這個(gè)在絕大多少情況下都是為真的。在Netty中只有寫半包的時(shí)候才需要注冊(cè)寫事件，如果一次寫就完全把數(shù)據(jù)寫入了緩沖區(qū)就不需要注冊(cè)寫事件。3.2 Tars客戶端發(fā)起請(qǐng)求到服務(wù)器的流程Communicator.stringToProxy()根據(jù)servantName等配置信息創(chuàng)建通信器。ServantProxyFactory.getServantProxy()調(diào)用工廠方法創(chuàng)建servant代理。ObjectProxyFactory.getObjectProxy()調(diào)用工廠方法創(chuàng)建obj代理。TarsProtocolInvoker.create()創(chuàng)建協(xié)議調(diào)用者。ServantProtocolInvoker.initClient(Url url)根據(jù)servantProxyConfig中的配置信息找到servant的ip端口等進(jìn)行初始化ServantClient。ClientPoolManager.getSelectorManager()如果第一次調(diào)用selectorManager是空的就會(huì)去初始化selectorManager。reactorSet = new Reactor[selectorPoolSize]; SelectorManager初始化構(gòu)造類中的會(huì)根據(jù)selectorPoolSize（默認(rèn)是2）的配置創(chuàng)建Reactor線程數(shù)組。線程名稱的前綴是servant-proxy-加上CommunicatorId，CommunicatorId生成規(guī)則是由locator的地址生成的UUID。啟動(dòng)reactor線程。3.3 Tars服務(wù)端啟動(dòng)步驟tars支持TCP和UDP兩種協(xié)議，RPC場(chǎng)景下是使用TCP協(xié)議。new SelectorManager()根據(jù)配置信息初始化selectorManager，線程池大小 processors > 8 ? 4 + (processors * 5 / 8) : processors + 1；線程名稱前綴是server-tcp-reactor，然后啟動(dòng)reactor線程數(shù)組中的所有線程。開(kāi)啟服務(wù)端監(jiān)聽(tīng)的ServerSocketChannel，綁定服務(wù)端本地ip和監(jiān)聽(tīng)的端口號(hào)，設(shè)置TCP連接請(qǐng)求隊(duì)列的最大容量為1024；設(shè)置非阻塞模式。選取reactor線程數(shù)組中第0個(gè)線程作為服務(wù)端監(jiān)聽(tīng)連接OP_ACCEPT就緒事件的線程。

代碼4：

public void bind(AppService appService) throws IOException {     // 此處略去非關(guān)鍵代碼     if (endpoint.type().equals("tcp")) {  // 1        this.selectorManager = new SelectorManager(Utils.getSelectorPoolSize(), new ServantProtocolFactory(codec), threadPool, processor, keepAlive, "server-tcp-reactor", false);     // 2        this.selectorManager.setTcpNoDelay(serverCfg.isTcpNoDelay());        this.selectorManager.start();        ServerSocketChannel serverChannel = ServerSocketChannel.open();        serverChannel.socket().bind(new InetSocketAddress(endpoint.host(), endpoint.port()), 1024);   // 3        serverChannel.configureBlocking(false);              selectorManager.getReactor(0).registerChannel(serverChannel, SelectionKey.OP_ACCEPT);  // 4    } else if (endpoint.type().equals("udp")) {        this.selectorManager = new SelectorManager(1, new ServantProtocolFactory(codec), threadPool, processor, false, "server-udp-reactor", true);        this.selectorManager.start();        // UDP開(kāi)啟的是DatagramChannel        DatagramChannel serverChannel = DatagramChannel.open();        DatagramSocket socket = serverChannel.socket();        socket.bind(new InetSocketAddress(endpoint.host(), endpoint.port()));        serverChannel.configureBlocking(false);        // UDP協(xié)議不需要建連，監(jiān)聽(tīng)的是OP_READ就緒事件        this.selectorManager.getReactor(0).registerChannel(serverChannel, SelectionKey.OP_READ);    }}

3.4 Reactor線程啟動(dòng)流程多路復(fù)用器開(kāi)始輪詢檢查 是否有就緒的事件。處理register隊(duì)列中剩余的channel注冊(cè)到當(dāng)前reactor線程的多路復(fù)用器selector中。獲取已選鍵集中所有就緒的channel。更新Session中最近操作時(shí)間，Tars服務(wù)端啟動(dòng)時(shí)會(huì)調(diào)用 startSessionManager() , 單線程每30s掃描一次session會(huì)話列表，會(huì)檢查每個(gè)session的 lastUpdateOperationTime 與當(dāng)前時(shí)間的時(shí)間差，如果超過(guò)60秒會(huì)將過(guò)期session對(duì)應(yīng)的channel踢除。分發(fā)IO事件進(jìn)行處理。處理unregister隊(duì)列中剩余的channel,從當(dāng)前reactor線程的多路復(fù)用器selector中解除注冊(cè)。

代碼5：

public void run() {        while (!Thread.interrupted()) {            selector.select();  // 1            processRegister();  // 2            Iterator iter = selector.selectedKeys().iterator();   //  3            while (iter.hasNext()) {                SelectionKey key = iter.next();                iter.remove();                if (!key.isValid()) continue;                try {                    if (key.attachment() != null && key.attachment() instanceof Session) {                      ((Session) key.attachment()).updateLastOperationTime(); //4                    }                 dispatchEvent(key);    // 5                } catch (Throwable ex) {                 disConnectWithException(key, ex);                }            }            processUnRegister();  // 6        }}

3.5IO事件分發(fā)處理

每個(gè)reactor線程都有一個(gè)專門的Accepter類去處理各種IO事件。TCPAccepter可以處理全部的四種事件（OP_ACCEPT、OP_CONNECT、OP_WRITE、OP_READ）、UDPAccepter由于不需要建立連接所以只需要處理讀和寫兩種事件。

1、處理OP_ACCEPT

獲取channel，處理TCP請(qǐng)求。為這個(gè)TCP請(qǐng)求創(chuàng)建TCPSession，會(huì)話的狀態(tài)是服務(wù)器已連接會(huì)話注冊(cè)到sessionManager中，Tars服務(wù)可配置最大連接數(shù)maxconns，如果超過(guò)就會(huì)關(guān)閉當(dāng)前會(huì)話。尋找下一個(gè)reactor線程進(jìn)行多路復(fù)用器與channel的綁定。

代碼6：

public void handleAcceptEvent(SelectionKey key) throws IOException {    ServerSocketChannel server = (ServerSocketChannel) key.channel(); // 1    SocketChannel channel = server.accept();       channel.socket().setTcpNoDelay(selectorManager.isTcpNoDelay());    channel.configureBlocking(false);    Utils.setQosFlag(channel.socket());    TCPSession session = new TCPSession(selectorManager);    // 2    session.setChannel(channel);    session.setStatus(SessionStatus.SERVER_CONNECTED);    session.setKeepAlive(selectorManager.isKeepAlive());    session.setTcpNoDelay(selectorManager.isTcpNoDelay());    SessionManager.getSessionManager().registerSession(session);   // 3      selectorManager.nextReactor().registerChannel(channel, SelectionKey.OP_READ, session); // 4}

2、處理OP_CONNECT

獲取客戶端連接過(guò)來(lái)的channel通道獲取Session與服務(wù)器建立連接，將關(guān)注的興趣OPS設(shè)置為ready就緒事件，session中的狀態(tài)修改為客戶端已連接

代碼7：

public void handleConnectEvent(SelectionKey key) throws IOException {    SocketChannel client = (SocketChannel) key.channel();  // 1    TCPSession session = (TCPSession) key.attachment();   //2    if (session == null) throw new RuntimeException("The session is null when connecting to ...");    try {  // 3        client.finishConnect();        key.interestOps(SelectionKey.OP_READ);        session.setStatus(SessionStatus.CLIENT_CONNECTED);    } finally {        session.finishConnect();    }}

3.處理OP_WRITE、處理OP_READ

調(diào)用session.read()和session.doWrite()方法處理讀寫事件

代碼8：

public void handleReadEvent(SelectionKey key) throws IOException {    TCPSession session = (TCPSession) key.attachment();    if (session == null) throw new RuntimeException("The session is null when reading data...");    session.read();}public void handleWriteEvent(SelectionKey key) throws IOException {    TCPSession session = (TCPSession) key.attachment();    if (session == null) throw new RuntimeException("The session is null when writing data...");    session.doWrite();}

3.6 seesion中網(wǎng)絡(luò)讀寫的事件詳細(xì)處理過(guò)程

1.讀事件處理

申請(qǐng)2k的ByteBuffer空間，讀取channel中的數(shù)據(jù)到readBuffer中。根據(jù)sessionStatus判斷是客戶端讀響應(yīng)還是服務(wù)器讀請(qǐng)求，分別進(jìn)行處理。

代碼9：

protected void read() throws IOException {    int ret = readChannel();    if (this.status == SessionStatus.CLIENT_CONNECTED) {        readResponse();    } else if (this.status == SessionStatus.SERVER_CONNECTED) {        readRequest();    } else {        throw new IllegalStateException("The current session status is invalid. [status:" + this.status + "]");    }    if (ret < 0) {        close();        return;    }}private int readChannel() throws IOException {    int readBytes = 0, ret = 0;    ByteBuffer data = ByteBuffer.allocate(1024 * 2);  // 1    if (readBuffer == null) {        readBuffer = IoBuffer.allocate(bufferSize);    }       // 2    while ((ret = ((SocketChannel) channel).read(data)) > 0) {        data.flip();  // 3        readBytes += data.remaining();        readBuffer.put(data.array(), data.position(), data.remaining());        data.clear();    }    return ret < 0 ? ret : readBytes;}

①客戶端讀響應(yīng)

從當(dāng)前readBuffer中的內(nèi)容復(fù)制到一個(gè)新的臨時(shí)buffer中，并且切換到讀模式，使用TarsCodec類解析出buffer內(nèi)的協(xié)議字段到response，WorkThread線程通知Ticket處理response。如果response為空，則重置tempBuffer到mark的位置，重新解析協(xié)議。

代碼10：

public void readResponse() {    Response response = null;    IoBuffer tempBuffer = null;        tempBuffer = readBuffer.duplicate().flip();        while (true) {            tempBuffer.mark();            if (tempBuffer.remaining() > 0) {                response = selectorManager.getProtocolFactory().getDecoder().decodeResponse(tempBuffer, this);            } else {                response = null;            }            if (response != null) {                if (response.getTicketNumber() == Ticket.DEFAULT_TICKET_NUMBER) response.setTicketNumber(response.getSession().hashCode());                selectorManager.getThreadPool().execute(new WorkThread(response, selectorManager));            } else {                tempBuffer.reset();                readBuffer = resetIoBuffer(tempBuffer);                break;            }        }}

②服務(wù)器讀請(qǐng)求

任務(wù)放入線程池交給 WorkThread線程，最終交給Processor類出構(gòu)建請(qǐng)求的響應(yīng)體，包括分布式上下文，然后經(jīng)過(guò)FilterChain的處理，最終通過(guò)jdk提供的反射方法invoke服務(wù)端本地的方法然后返回response。如果線程池拋出拒絕異常，則返回SERVEROVERLOAD = -9，服務(wù)端過(guò)載保護(hù)。如果request為空，則重置tempBuffer到mark的位置，重新解析協(xié)議。

代碼11：

public void readRequest() {    Request request = null;    IoBuffer tempBuffer = readBuffer.duplicate().flip();        while (true) {            tempBuffer.mark();            if (tempBuffer.remaining() > 0) {                request = selectorManager.getProtocolFactory().getDecoder().decodeRequest(tempBuffer, this);            } else {                request = null;            }            if (request != null) {                try {                    request.resetBornTime();                    selectorManager.getThreadPool().execute(new WorkThread(request, selectorManager));                } catch (RejectedExecutionException e) {                  selectorManager.getProcessor().overload(request, request.getIoSession());                } catch (Exception ex) {                  ex.printStackTrace();                }            } else {                    tempBuffer.reset();                readBuffer = resetIoBuffer(tempBuffer);                break;            }        }}

2.寫事件處理

同樣也包括客戶端寫請(qǐng)求和服務(wù)端寫響應(yīng)兩種，其實(shí)這兩種都是往TCPSession中的LinkedBlockingQueue（有界隊(duì)列最大8K）中插入ByteBuffer。LinkedBlockingQueue中的ByteBuffer最終會(huì)由TCPAcceptor中的handleWriteEvent監(jiān)聽(tīng)寫就緒事件并消費(fèi)。

代碼12：

protected void write(IoBuffer buffer) throws IOException {    if (buffer == null) return;    if (channel == null || key == null) throw new IOException("Connection is closed");    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();    }}

四、總結(jié)

本文主要介紹了Java NIO編程的基礎(chǔ)知識(shí) 和 Tars-Java 1.7.2版本的網(wǎng)絡(luò)編程模塊的源碼實(shí)現(xiàn)。

在最新的Tars-Java的master分支中我們可以發(fā)現(xiàn)網(wǎng)絡(luò)編程已經(jīng)由NIO改成了Netty，雖然Netty更加成熟穩(wěn)定，但是作為學(xué)習(xí)者了解NIO的原理也是掌握網(wǎng)絡(luò)編程的必經(jīng)之路。

關(guān)鍵詞：