版權聲明：本文引用https://yinwj.blog.csdn.net/article/details/48274255

從這篇博文開始，我們將進入一個新文章系列。這個文章系列專門整理總結了目前系統(tǒng)間通信的主要原理、手段和實現(xiàn)。我們將講解典型的信息格式、講解傳統(tǒng)的RMI調用并延伸出來重點講解RPC調用和使用案例；最后我們還會講到SOA架構的實現(xiàn)，包括ESB實現(xiàn)和服務注冊/治理的實現(xiàn)，同樣包括原理、實現(xiàn)和使用案例。
系統(tǒng)間通信是架構師需要掌握的又一個關鍵技術領域，如果說理解和掌握負載均衡層技術需要您有一定的linux系統(tǒng)知識和操作系統(tǒng)知識的話，那么理解和掌握系統(tǒng)間通信層技術，需要您有一定的編程經(jīng)驗（最好是JAVA編程經(jīng)驗，因為我們會主要以JAVA技術作為實例演示）。

1. 聊天場景

首先我們來看一個顯示場景：在現(xiàn)實生活中有兩個人技術人員A和B，在進行一問一答形式的交流。如下圖所示：

我們來看這幅圖的中的幾個要點：

• 他們兩都使用中文進行交流。如果他們一人使用的是南斯拉夫語另一人使用的是索馬里語，并且相互都不能理解對方的語系，很顯然A所要表達的內容B是無法理解的。
• 他們的聲音是在空氣中進行傳播的。空氣除了支撐他們的呼吸外，還支撐了他們聲音的傳播。如果沒有空氣他們是無法知道對方用中文說了什么。
• 他們的交流方式是協(xié)調一致的，即A問完一個問題后，等待B進行回答。收到B的回答后，A才能問下一個問題。
• 由于都是人類，所以他們處理信息的方式也是一樣的：用嘴說話，用耳朵聽話，用大腦處理形成結果。
• 目前這個交流場景下，只有A和B兩個人。但是隨時有可能增加N個人進來。第N個人可能不是采用中文進行交流。

2. 信息格式

很明顯通過中文的交談，兩個人相互明白了對方的意圖。為了保證信息傳遞的高效性，我們一定會將信息做成某種參與者都理解的格式。例如：中文有其特定的語法結構，例如主謂賓，定狀補。

在計算機領域為了保證信息能夠被處理，信息也會被做成特定的格式，而且要確保目標能夠明白這種格式。常用的信息格式包括：

2.1 XML

可擴展標記語言，這個語言由W3C（萬維網(wǎng)聯(lián)盟）進行發(fā)布和維護。XML語言應用之廣泛，擴展之豐富。適合做網(wǎng)絡通信的信息描述格式（一般是“應用層”協(xié)議了）。例如Google 定義的XMPP通信協(xié)議就是使用XML進行描述的；不過XML的更廣泛使用場景是對系統(tǒng)環(huán)境進行描述（因為它會造成較多的不必要的內容傳輸），例如服務器的配置描述、Spring的配置描述、Maven倉庫描述等等。

2.2 JSON

JSON(JavaScript Object Notation) 是一種輕量級的數(shù)據(jù)交換格式。它和XML的設計思路是一致的：和語言無關（流行的語言都支持JSON格式描述：Go、Python、C、C++、C#、JAVA、Erlang、JavaScript等等）；但是和XML不同，JSON的設計目標就是為了進行通信。要描述同樣的數(shù)據(jù)，JSON格式的容量會更小。

2.3 protocol buffer

protocol buffer(以下簡稱PB)是google 的一種數(shù)據(jù)交換的格式，它獨立于語言，獨立于平臺。google 提供了三種語言的實現(xiàn)：java、c++ 和 python，每一種實現(xiàn)都包含了相應語言的編譯器以及庫文件。

2.4 TLV

三元組編碼，T（標記/類型域）L（長度/大小域）V（值/內容域），通常這種信息格式用于金融、軍事領域。它通過字節(jié)的位運算來進行信息的序列化/反序列化（據(jù)說微信的信息格式也采用的是TLV，但實際情況我不清楚）：

這里有一篇介紹TLV的文章：《通信協(xié)議之序列化TLV》，TLV格式所攜帶的內容是最有效的，它就連JSON中用于分割層次的“{}”符號都沒有。

2.5 自定義

當然，如果您的兩個內部系統(tǒng)已經(jīng)約定好了一種信息格式，您當然可以使用自己定制的格式進行描述。您可以使用C++描述一個結構體，然后序列化/反序列它，或者使用一個純文本，以“|”號分割這些字符串，然后序列化/反序列它。

在這個系列的博文中，我們不會把信息格式作為一個重點，但是會花一些篇幅去比較各種信息格式在網(wǎng)絡上傳輸?shù)乃俣取⑿阅?#xff0c;并為大家介紹幾種典型的信息格式選型場景。

3. 網(wǎng)絡協(xié)議

如文中第一張圖描述的場景，有一個我們看不到但是卻很重要的元素：空氣。聲音在空氣中完成傳播，真空無法傳播聲音。同樣信息是在網(wǎng)絡中完成傳播的，沒有網(wǎng)絡就沒法傳播信息。網(wǎng)絡協(xié)議就是計算機領域的“空氣”，下圖中我們以OSI模型作為參考：

• 物理層：物理層就是我們的網(wǎng)絡設備層，例如我們的網(wǎng)卡、交換機等設備，在他們之間我們一般傳遞的是電信號或者光信號。
• 數(shù)據(jù)鏈路層：數(shù)據(jù)鏈路又分為物理鏈路和邏輯鏈路。物理鏈路負責組合一組電信號，稱之為“幀”；邏輯鏈路層通過一些規(guī)則和協(xié)議保證幀傳輸?shù)恼_性，并且可以使來自于多個源/目標 的幀在同一個物理鏈路上進行傳輸，實現(xiàn)“鏈路復用”。
• 網(wǎng)絡層：網(wǎng)絡層使用最廣泛的協(xié)議是IP協(xié)議（又分為IPV4協(xié)議和IPV6協(xié)議），IPX協(xié)議。這些協(xié)議解決的是源和目標的定位問題，以及從源如何到達目標的問題。
• 傳輸層：TCP、UDP是傳輸層最常使用的協(xié)議，傳輸層的最重要工作就是攜帶內容信息了，并且通過他們的協(xié)議規(guī)范提供某種通信機制。舉例來說，TCP協(xié)議中的通信機制是：首先進行三次通信握手，然后再進行正式數(shù)據(jù)的傳送，并且通過校驗機制保證每個數(shù)據(jù)報文的正確性，如果數(shù)據(jù)報文錯誤了，則重新發(fā)送。
• 應用層：HTTP協(xié)議、FTP協(xié)議、TELNET協(xié)議這些都是應用層協(xié)議。應用層協(xié)議是最靈活的協(xié)議，甚至可以由程序員自行定義應用層協(xié)議。下圖我們表示了HTTP協(xié)議的工作方式：
  
  在這個系列的博文中，我們不會把網(wǎng)絡協(xié)議作為一個重點。這是因為網(wǎng)絡網(wǎng)絡協(xié)議的知識是一個相對獨立的的知識領域，十幾篇文章都不一定講得清楚。如果您對網(wǎng)絡協(xié)議有興趣，這里推薦兩本書：《TCP/IP詳解.卷1-協(xié)議》和《TCP/IP詳解.卷2-實現(xiàn)》。

4. 通信方式|框架

在文章最前面我們看到其中一個人規(guī)定了一種溝通方式：“你必須把我說的話聽完，然后給我反饋后。我才會問第二個問題”。這種溝通方式雖然溝通效率不高，但是很有效：一個問題一個問題的處理。

但是如果參與溝通的人處理信息的能力比較強，那么他們還可以采用另一種溝通方式：“我給我提的問題編了一個號，在問完第X個問題后，我不會等待你返回，就會問第X+1個問題，同樣你在聽完我第X個問題后，一邊處理我的問題，一邊聽我第X+1個問題。”

實際上以上兩種現(xiàn)實中的溝通方式，在計算機領域是可以找到對應的通信方式的，這就是我們這個系列的博文會著重講的BIO（阻塞模式）通信和NIO（非阻塞模式）。

4.1 BIO通信模式

以前大多數(shù)網(wǎng)絡通信方式都是阻塞模式的，即：

客戶端向服務器端發(fā)出請求后，客戶端會一直等待（不會再做其他事情），直到服務器端返回結果或者網(wǎng)絡出現(xiàn)問題。

服務器端同樣的，當在處理某個客戶端A發(fā)來的請求時，另一個客戶端B發(fā)來的請求會等待，直到服務器端的這個處理線程完成上一個處理。

如下圖所示：

傳統(tǒng)的BIO通信方式存在幾個問題：

同一時間，服務器只能接受來自于客戶端A的請求信息；雖然客戶端A和客戶端B的請求是同時進行的，但客戶端B發(fā)送的請求信息只能等到服務器接受完A的請求數(shù)據(jù)后，才能被接受。

由于服務器一次只能處理一個客戶端請求，當處理完成并返回后（或者異常時），才能進行第二次請求的處理。很顯然，這樣的處理方式在高并發(fā)的情況下，是不能采用的。

上面說的情況是服務器只有一個線程的情況，那么讀者會直接提出我們可以使用多線程技術來解決這個問題：

當服務器收到客戶端X的請求后，（讀取到所有請求數(shù)據(jù)后）將這個請求送入一個獨立線程進行處理，然后主線程繼續(xù)接受客戶端Y的請求。

客戶端一側，也可以使用一個子線程和服務器端進行通信。這樣客戶端主線程的其他工作就不受影響了，當服務器端有響應信息的時候再由這個子線程通過 監(jiān)聽模式/觀察模式（等其他設計模式）通知主線程。

如下圖所示：

但是使用線程來解決這個問題實際上是有局限性的：

• 雖然在服務器端，請求的處理交給了一個獨立線程進行，但是操作系統(tǒng)通知accept()的方式還是單個的。也就是，實際上是服務器接收到數(shù)據(jù)報文后的“業(yè)務處理過程”可以多線程，但是數(shù)據(jù)報文的接受還是需要一個一個的來（下文的示例代碼和debug過程我們可以明確看到這一點）
• 在linux系統(tǒng)中，可以創(chuàng)建的線程是有限的。我們可以通過cat /proc/sys/kernel/threads-max 命令查看可以創(chuàng)建的最大線程數(shù)。當然這個值是可以更改的，但是線程越多，CPU切換所需的時間也就越長，用來處理真正業(yè)務的需求也就越少。
• 創(chuàng)建一個線程是有較大的資源消耗的。JVM創(chuàng)建一個線程的時候，即使這個線程不做任何的工作，JVM都會分配一個堆棧空間。這個空間的大小默認為128K，您可以通過-Xss參數(shù)進行調整。
• 當然您還可以使用ThreadPoolExecutor線程池來緩解線程的創(chuàng)建問題，但是又會造成BlockingQueue積壓任務的持續(xù)增加，同樣消耗了大量資源。另外，如果您的應用程序大量使用長連接的話，線程是不會關閉的。這樣系統(tǒng)資源的消耗更容易失控。

那么，如果你真想單純使用線程解決阻塞的問題，那么您自己都可以算出來您一個服務器節(jié)點可以一次接受多大的并發(fā)了。看來，單純使用線程解決這個問題不是最好的辦法。

4.2 BIO通信方式深入分析

在這個系列的博文中，通信方式/框架將作為一個重點進行講解。包括NIO的原理，并通過講解Netty的使用、JAVA原生NIO框架的使用，去熟悉這些核心原理。

實際上從上文中我們可以看出，BIO的問題關鍵不在于是否使用了多線程（包括線程池）處理這次請求，而在于accept()、read()的操作點都是被阻塞。要測試這個問題，也很簡單。我們模擬了20個客戶端（用20根線程模擬），利用JAVA的同步計數(shù)器CountDownLatch，保證這20個客戶都初始化完成后然后同時向服務器發(fā)送請求，然后我們來觀察一下Server這邊接受信息的情況。

4.2.1 模擬20個客戶端并發(fā)請求，服務器端使用單線程：

• 客戶端代碼（SocketClientDaemon）

package testBSocket;import java.util.concurrent.CountDownLatch;public class SocketClientDaemon {public static void main(String[] args) throws Exception {Integer clientNumber = 20;CountDownLatch countDownLatch = new CountDownLatch(clientNumber);//分別開始啟動這20個客戶端for(int index = 0 ; index < clientNumber ; index++ , countDownLatch.countDown()) {SocketClientRequestThread client = new SocketClientRequestThread(countDownLatch, index);new Thread(client).start();}//這個wait不涉及到具體的實驗邏輯，只是為了保證守護線程在啟動所有線程后，進入等待狀態(tài)synchronized (SocketClientDaemon.class) {SocketClientDaemon.class.wait();}} }

• 客戶端代碼（SocketClientRequestThread模擬請求）

package testBSocket;import java.io.IOException; import java.io.InputStream; import java.io.OutputStream; import java.net.Socket; import java.util.concurrent.CountDownLatch;import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.log4j.BasicConfigurator;/*** 一個SocketClientRequestThread線程模擬一個客戶端請求。* @author yinwenjie*/ public class SocketClientRequestThread implements Runnable {static {BasicConfigurator.configure();}/*** 日志*/private static final Log LOGGER = LogFactory.getLog(SocketClientRequestThread.class);private CountDownLatch countDownLatch;/*** 這個線層的編號* @param countDownLatch*/private Integer clientIndex;/*** countDownLatch是java提供的同步計數(shù)器。* 當計數(shù)器數(shù)值減為0時，所有受其影響而等待的線程將會被激活。這樣保證模擬并發(fā)請求的真實性* @param countDownLatch*/public SocketClientRequestThread(CountDownLatch countDownLatch , Integer clientIndex) {this.countDownLatch = countDownLatch;this.clientIndex = clientIndex;}@Overridepublic void run() {Socket socket = null;OutputStream clientRequest = null;InputStream clientResponse = null;try {socket = new Socket("localhost",83);clientRequest = socket.getOutputStream();clientResponse = socket.getInputStream();//等待，直到SocketClientDaemon完成所有線程的啟動，然后所有線程一起發(fā)送請求this.countDownLatch.await();//發(fā)送請求信息clientRequest.write(("這是第" + this.clientIndex + " 個客戶端的請求。").getBytes());clientRequest.flush();//在這里等待，直到服務器返回信息SocketClientRequestThread.LOGGER.info("第" + this.clientIndex + "個客戶端的請求發(fā)送完成，等待服務器返回信息");int maxLen = 1024;byte[] contextBytes = new byte[maxLen];int realLen;String message = "";//程序執(zhí)行到這里，會一直等待服務器返回信息（注意，前提是in和out都不能close，如果close了就收不到服務器的反饋了）while((realLen = clientResponse.read(contextBytes, 0, maxLen)) != -1) {message += new String(contextBytes , 0 , realLen);}SocketClientRequestThread.LOGGER.info("接收到來自服務器的信息:" + message);} catch (Exception e) {SocketClientRequestThread.LOGGER.error(e.getMessage(), e);} finally {try {if(clientRequest != null) {clientRequest.close();}if(clientResponse != null) {clientResponse.close();}} catch (IOException e) {SocketClientRequestThread.LOGGER.error(e.getMessage(), e);}}} }

• 服務器端（SocketServer1）單個線程

package testBSocket;import java.io.InputStream; import java.io.OutputStream; import java.net.ServerSocket; import java.net.Socket;import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.log4j.BasicConfigurator;public class SocketServer1 {static {BasicConfigurator.configure();}/*** 日志*/private static final Log LOGGER = LogFactory.getLog(SocketServer1.class);public static void main(String[] args) throws Exception{ServerSocket serverSocket = new ServerSocket(83);try {while(true) {Socket socket = serverSocket.accept();//下面我們收取信息InputStream in = socket.getInputStream();OutputStream out = socket.getOutputStream();Integer sourcePort = socket.getPort();int maxLen = 2048;byte[] contextBytes = new byte[maxLen];//這里也會被阻塞，直到有數(shù)據(jù)準備好int realLen = in.read(contextBytes, 0, maxLen);//讀取信息String message = new String(contextBytes , 0 , realLen);//下面打印信息SocketServer1.LOGGER.info("服務器收到來自于端口：" + sourcePort + "的信息：" + message);//下面開始發(fā)送信息out.write("回發(fā)響應信息！".getBytes());//關閉out.close();in.close();socket.close();}} catch(Exception e) {SocketServer1.LOGGER.error(e.getMessage(), e);} finally {if(serverSocket != null) {serverSocket.close();}}} }

4.2.2 使用多線程來優(yōu)化服務器端的處理過程

客戶端代碼和上文一樣，最主要是更改服務器端的代碼：

package testBSocket;import java.io.IOException; import java.io.InputStream; import java.io.OutputStream; import java.net.ServerSocket; import java.net.Socket;import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.log4j.BasicConfigurator;public class SocketServer2 {static {BasicConfigurator.configure();}private static final Log LOGGER = LogFactory.getLog(SocketServer2.class);public static void main(String[] args) throws Exception{ServerSocket serverSocket = new ServerSocket(83);try {while(true) {Socket socket = serverSocket.accept();//當然業(yè)務處理過程可以交給一個線程（這里可以使用線程池）,并且線程的創(chuàng)建是很耗資源的。//最終改變不了.accept()只能一個一個接受socket的情況,并且被阻塞的情況SocketServerThread socketServerThread = new SocketServerThread(socket);new Thread(socketServerThread).start();}} catch(Exception e) {SocketServer2.LOGGER.error(e.getMessage(), e);} finally {if(serverSocket != null) {serverSocket.close();}}} }/*** 當然，接收到客戶端的socket后，業(yè)務的處理過程可以交給一個線程來做。* 但還是改變不了socket被一個一個的做accept()的情況。* @author yinwenjie*/ class SocketServerThread implements Runnable {/*** 日志*/private static final Log LOGGER = LogFactory.getLog(SocketServerThread.class);private Socket socket;public SocketServerThread (Socket socket) {this.socket = socket;}@Overridepublic void run() {InputStream in = null;OutputStream out = null;try {//下面我們收取信息in = socket.getInputStream();out = socket.getOutputStream();Integer sourcePort = socket.getPort();int maxLen = 1024;byte[] contextBytes = new byte[maxLen];//使用線程，同樣無法解決read方法的阻塞問題，//也就是說read方法處同樣會被阻塞，直到操作系統(tǒng)有數(shù)據(jù)準備好int realLen = in.read(contextBytes, 0, maxLen);//讀取信息String message = new String(contextBytes , 0 , realLen);//下面打印信息SocketServerThread.LOGGER.info("服務器收到來自于端口：" + sourcePort + "的信息：" + message);//下面開始發(fā)送信息out.write("回發(fā)響應信息！".getBytes());} catch(Exception e) {SocketServerThread.LOGGER.error(e.getMessage(), e);} finally {//試圖關閉try {if(in != null) {in.close();}if(out != null) {out.close();}if(this.socket != null) {this.socket.close();}} catch (IOException e) {SocketServerThread.LOGGER.error(e.getMessage(), e);}}} }

4.2.3 服務器端的執(zhí)行效果

我相信服務器使用單線程的效果就不用看了，我們主要看一看服務器使用多線程處理時的情況：

4.2.4

那么重點的問題并不是“是否使用了多線程”，而是為什么accept()、read()方法會被阻塞。即：異步IO模式 就是為了解決這樣的并發(fā)性存在的。但是為了說清楚異步IO模式，在介紹IO模式的時候，我們就要首先了解清楚，什么是 阻塞式同步、非阻塞式同步、多路復用同步模式。
API文檔中對于 serverSocket.accept() 方法的使用描述：

Listens for a connection to be made to this socket and accepts it. The method blocks until a connection is made.

那么我們首先來看看為什么serverSocket.accept()會被阻塞。這里涉及到阻塞式同步IO的工作原理：

• 服務器線程發(fā)起一個accept動作，詢問操作系統(tǒng) 是否有新的socket套接字信息從端口X發(fā)送過來。
  
• 注意，是詢問操作系統(tǒng)。也就是說socket套接字的IO模式支持是基于操作系統(tǒng)的，那么自然同步IO/異步IO的支持就是需要操作系統(tǒng)級別的了。如下圖：
  
• 如果操作系統(tǒng)沒有發(fā)現(xiàn)有套接字從指定的端口X來，那么操作系統(tǒng)就會等待。這樣serverSocket.accept()方法就會一直等待。這就是為什么accept()方法為什么會阻塞：它內部的實現(xiàn)是使用的操作系統(tǒng)級別的同步IO。

阻塞IO 和 非阻塞IO 這兩個概念是程序級別的。主要描述的是程序請求操作系統(tǒng)IO操作后，如果IO資源沒有準備好，那么程序該如何處理的問題：前者等待；后者繼續(xù)執(zhí)行（并且使用線程一直輪詢，直到有IO資源準備好了）

同步IO 和 非同步IO，這兩個概念是操作系統(tǒng)級別的。主要描述的是操作系統(tǒng)在收到程序請求IO操作后，如果IO資源沒有準備好，該如何相應程序的問題：前者不響應，直到IO資源準備好以后；后者返回一個標記（好讓程序和自己知道以后的數(shù)據(jù)往哪里通知），當IO資源準備好以后，再用事件機制返回給程序。

與50位技術專家面對面20年技術見證，附贈技術全景圖

總結

以上是生活随笔為你收集整理的系统间通信1：阻塞与非阻塞式通信A的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內容還不錯，歡迎將生活随笔推薦給好友。