Kafka網絡模型基礎-Reactor（上）

上一篇我們介紹了 Kafka 生產者的消息管理模型，對於消息的管理，可以理解為 Kafka 生產者組裝消息內容，準備批量發送到 Broker 端，這是一個 Ready 的過程。那麼在我們介紹Kafka的網絡模型之前，我們先說一下常用的網絡模型--Reactor。

BIO網絡模型

網絡編程的基本模型是C/S模型，也就是兩個進程之間相互通信，其中服務端提供位置信息（IP和監聽端口）。客戶端通過 connect 操作向服務端監聽端口發送連接請求，通過TCP 3次握手之後，如果連接成功，雙方就可以通過 Socket 進行通信。

BIO通信模型-新建線程

從上面可以看出，傳統的BIO網絡模型中，通常由一個 Acceptor 來接收請求。收到客戶端的請求後，會為每個請求創建一個線程進行鏈路處理，處理完成之後會通過輸出流寫會客戶端，然後線程銷燬。下面我們看一下用Java實現的BIO通信模型，首先是Server 端的實現。

public class TimeServer {
 public static void main(String[] args) throws Exception {
 int port = 8080;
 if (args != null && args.length > 0) {
 try {
 port = Integer.valueOf(args[0]);
 } catch (Exception e) {
 }
 }
 ServerSocket server = null;
 try {
 server = new ServerSocket(port);
 System.out.println("The time server is start in port : " + port);
 Socket socket = null;
 while (true) {
 socket = server.accept();
 new Thread(new TimeServerHandler(socket)).start();
 }
 } finally {
 if (server != null) {
 System.out.println("The time server is close");
 server.close();
 server = null;
 }
 }
 }
}

然後是 Server 端處理請求的線程，線程中持有 Socket 的句柄。

public class TimeServerHandler implements Runnable {
 private Socket socket;
 public TimeServerHandler(Socket socket) {
 this.socket = socket;
 }
 @Override
 public void run() {
 BufferedReader in = null;
 PrintWriter out = null;
 try {
 in = new BufferedReader(new InputStreamReader(this.socket.getInputStream()));
 out = new PrintWriter(this.socket.getOutputStream(), true);
 String currentTime = null;
 String body = null;
 while (true) {
 body = in.readLine();
 if (body == null) {
 break;
 }
 System.out.println("The time server receive order : " + body);
 currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body) ? new Date(System.currentTimeMillis()).toString() : "BAD ORDER";
 out.println(currentTime);
 }
 } catch (Exception e) {
 if (in != null) {
 try {
 in.close();
 } catch (Exception ex) {
 ex.printStackTrace();
 }
 }
 if (out != null) {
 out.close();
 out = null;
 }
 if (this.socket != null) {
 try {
 this.socket.close();
 } catch (Exception e1) {
 e1.printStackTrace();
 }
 this.socket = null;
 }
 }
 }
}

最後我們看一下 Client 端的實現。

public class TimeClient {
 public static void main(String[] args) {
 int port = 8080;
 if (args != null && args.length > 0) {
 try {
 port = Integer.valueOf(args[0]);
 } catch (Exception e) {
 }
 }
 Socket socket = null;
 BufferedReader in = null;
 PrintWriter out = null;
 try {
 socket = new Socket("127.0.0.1", port);
 in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
 out = new PrintWriter(socket.getOutputStream(), true);
 out.println("QUERY TIME ORDER");
 System.out.println("Send order 2 server succeed.");
 String resp = in.readLine();
 System.out.println("Now is : " + resp);
 } catch (Exception e) {
 } finally {
 if (out != null) {
 out.close();
 out = null;
 }
 if (in != null) {
 try {
 in.close();
 } catch (Exception e1){
 e1.printStackTrace();
 }
 }
 if (socket != null) {
 try {
 socket.close();
 } catch (Exception e2) {
 e2.printStackTrace();
 }
 socket = null;
 }
 }
 }
}

BIO通信模型-線程池

對於上面BIO模型新建線程的模式的優化場景是，可以將Server 處理線程使用線程池的方式去實現，這樣可以防止頻繁的創建和銷燬線程，可以從一定程度上面減少系統開銷。其模型圖如下所示：

Reactor模式（NIO編程）

對於Reactor模式主要有以下關鍵點：（1）是一種事件驅動模型；（2）可以處理多個輸入；（3）使用多路複用器將事件分發給對應的Handler處理。其類圖示意如下：

（1）Event_handler：事件處理器，可以根據事件的不同狀態創建不同狀態的處理器。

（2）Handle：可以理解為事件，在網絡編程中對應 Socket。

（3）Reactor（Dispatcher）：反應器/分發器，用於管理Event_handler，分發Event 的容器，也是一個事件處理調度器。Tomcat 的Dispatcher 就是一個很好的實現，用於接收到網絡請求後，分發給響應的處理器去處理。

（4）Demultiplexer：多路複用器，阻塞等待Handle事件的到來。Java NIO 中使用 Selector 來封裝，底層由操作系統提供實現，如Linux 的 epoll()。

下面我們看一下 Reactor 模式的執行流程：

（1）首先初始化 Reactor，並初始化一個Handle 到Event_handle 的 Map 中。

（2）註冊 Event_handle 到Reactor 中，每個Event_handle 包含對Handle 的引用，從而建立 Handle 到 Event_handle 的映射Map。

（3）調用 Reactor 中的handle_events() 啟動 Event Loop，在循環中調用select()方法獲取就緒的 Event 列表。

（4）當 Event 發生後，select() 阻塞方法就會返回，此時 Reactor 會根據 Handle 找到註冊的 Event_handle，並回調 Event_handle 的 handle_events() 方法。

（5）在 Event_handle 的 handle_events() 方法中，還可以向 Reactor 註冊新的Event_handle，比如連接成功後，向 Reactor 註冊 寫事件。

Reactor 手工實現

下面我們手工實現一下上面介紹的Reactor 模型，其交互模式圖如下所示：

InputSource：外部輸入類，表示要處理的對象。

Event：對 InputSource 的封裝，是Reactor 直接操作的對象。

EventType：表示Event 的類型。

EventHandler：處理事件的抽象類，封裝了不同事件處理器的公共邏輯和公共變量。

AcceptEventHandler/ReadEventHandler：EventHandler 的子類，表示具體事件的處理類。

Dispatcher：事件分發器，Reactor 模型解決的問題就是接收到任務後，快速的分配給不同的處理器處理。

Selector：輪詢選擇器，實現了輪詢隊列的事件狀態，可以處理當前能夠處理的事件。

Acceptor：事件接收類，負責初始化Selector 和接收緩衝隊列。

Server：服務入口，負責啟動 Reactor 和 Acceptor。

上面所說提及的具體代碼實現如下：

InputSource

/**
 * 外部輸入類, 用來表示需要reactor去處理的原始對象
 */
public class InputSource {
 private Object data;
 private long id;
 public InputSource(Object data, long id) {
 this.data = data;
 this.id = id;
 }
 @Override
 public String toString() {
 return "InputSource{" + "data=" + data + ", id=" + id + '}';
 }
}

Event

/**
 * reactor 模式的事件類, 可以理解為將輸入原始對象根據不同狀態包裝成一個事件類, reactor 模式裡處理的 event 事件對象
 */
public class Event {
 private InputSource source;
 private EventType type;
 public InputSource getSource() {
 return source;
 }
 public void setSource(InputSource source) {
 this.source = source;
 }
 public EventType getType() {
 return type;
 }
 public void setType(EventType type) {
 this.type = type;
 }
}

EventType

/**
 * 枚舉類型表示事件的不同類型
 */
public enum EventType {
 ACCEPT,
 READ,
 WRITE;
}

EventHandler

/**
 * 處理事件的抽象類，裡面包含了不同事件處理器的公共邏輯和公共對象
 */
public abstract class EventHandler {
 private InputSource source;
 public abstract void handle(Event event);
 public InputSource getSource() {
 return source;
 }
 public void setSource(InputSource source) {
 this.source = source;
 }
}

AcceptEventHandler

/**
 * ACCEPT事件處理器
 */
public class AcceptEventHandler extends EventHandler {
 private Selector selector;
 public AcceptEventHandler(Selector selector) {
 this.selector = selector;
 }
 @Override
 public void handle(Event event) {
 // 處理Accept的event事件
 if (event.getType() == EventType.ACCEPT) {
 // TODO 處理 ACCEPT 狀態的事件
 // 將事件狀態改為下一個READ狀態，並放入selector的緩衝隊列中
 Event readEvent = new Event();
 readEvent.setSource(event.getSource());
 readEvent.setType(EventType.READ);
 selector.addEvent(readEvent);
 }
 }
}

Dispatcher

/**
 * reactor模式中Dispatcher類，負責event的分發和eventHandler的維護
 */
public class Dispatcher {
 /**
 * 通過ConcurrentHashMap來維護不同事件處理器
 */
 Map eventHandlerMap = new ConcurrentHashMap<>();
 /**
 * 本例只維護一個selector負責事件選擇，netty為了保證性能實現了多個selector來保證循環處理性能，不同事件加入不同的selector的事件緩衝隊列
 */
 Selector selector;
 Dispatcher(Selector selector) {
 this.selector = selector;
 }
 /**
 * 在Dispatcher中註冊eventHandler
 */
 public void registEventHandler(EventType eventType, EventHandler eventHandler) {
 eventHandlerMap.put(eventType, eventHandler);
 }
 public void removeEventHandler(EventType eventType) {
 eventHandlerMap.remove(eventType);
 }
 public void handleEvents() {
 dispatch();
 }
 /**
 * 此例只是實現了簡單的事件分發給相應的處理器處理，例子中的處理器都是同步，在reactor模式的典型實現NIO中都是在handle異步處理，來保證非阻塞
 */
 private void dispatch() {
 while (true) {
 List events = selector.select();
 for (Event event : events) {
 EventHandler eventHandler = eventHandlerMap.get(event.getType());
 eventHandler.handle(event);
 }
 }
 }
}

Selector

/**
 * 事件輪循選擇器, selector主要實現了輪循隊列中的事件狀態, 取出當前能夠處理的狀態
 */
public class Selector {
 /** 定義一個鏈表阻塞queue實現緩衝隊列，用於保證線程安全 */
 private BlockingQueue eventQueue = new LinkedBlockingQueue 
<>();
 /** 定義一個object用於synchronize方法塊上鎖 */
 private Object lock = new Object();
 List select() {
 return select(0);
 }
 List select(long timeout) {
 if (timeout > 0) {
 if (eventQueue.isEmpty()) {
 synchronized (lock) {
 if (eventQueue.isEmpty()) {
 try {
 lock.wait(timeout);
 } catch (InterruptedException e) {
 }
 }
 }
 }
 }
 // TODO 例子中只是簡單的將event列表全部返回，可以在此處增加業務邏輯，選出符合條件的event進行返回
 List events = new ArrayList<>();
 eventQueue.drainTo(events);
 return events;
 }
 public void addEvent(Event e) {
 // 將event事件加入隊列
 boolean success = eventQueue.offer(e);
 if (success) {
 synchronized (lock) {
 // 如果有新增事件則對lock對象解鎖
 lock.notify();
 }
 }
 }
}

Acceptor

/**
 * reactor的事件接收類，負責初始化selector和接收緩衝隊列
 */
public class Acceptor implements Runnable {
 private int port;
 private Selector selector;
 /**
 * 代表 serversocket, 通過LinkedBlockingQueue來模擬外部輸入請求隊列
 */
 private BlockingQueue sourceQueue = new LinkedBlockingQueue<>();
 Acceptor(Selector selector, int port) {
 this.selector = selector;
 this.port = port;
 }
 /**
 * 外部有輸入請求後，需要加入到請求隊列中
 */
 public void addNewConnection(InputSource source) {
 sourceQueue.offer(source);
 }
 public int getPort() {
 return this.port;
 }
 @Override
 public void run() {
 while (true) {
 InputSource source = null;
 try {
 // 相當於 serversocket.accept()，接收輸入請求，該例從請求隊列中獲取輸入請求
 source = sourceQueue.take();
 } catch (InterruptedException e) {
 // ignore it;
 }
 // 接收到InputSource後將接收到event設置type為ACCEPT，並將source賦值給event
 if (source != null) {
 Event acceptEvent = new Event();
 acceptEvent.setSource(source);
 acceptEvent.setType(EventType.ACCEPT);
 selector.addEvent(acceptEvent);
 }
 }
 }
}

Server

public class Server {
 Selector selector = new Selector();
 Dispatcher eventLooper = new Dispatcher(selector);
 Acceptor acceptor;
 Server(int port) {
 acceptor = new Acceptor(selector, port);
 }
 public void start() {
 eventLooper.registEventHandler(EventType.ACCEPT, new AcceptEventHandler(selector));
 new Thread(acceptor, "Acceptor-" + acceptor.getPort()).start();
 eventLooper.handleEvents();
 }
}

參考：《Netty權威指南》、https://www.jianshu.com/p/188ef8462100