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