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<EventType, EventHandler> 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<Event> events = selector.select();

            for (Event event : events) {
                EventHandler eventHandler = eventHandlerMap.get(event.getType());
                eventHandler.handle(event);
            }
        }
    }

}

Selector

/**
 * 事件轮循选择器, selector主要实现了轮循队列中的事件状态, 取出当前能够处理的状态
 */
public class Selector {

    /** 定义一个链表阻塞queue实现缓冲队列，用于保证线程安全 */
    private BlockingQueue<Event> eventQueue = new LinkedBlockingQueue<>();
    /** 定义一个object用于synchronize方法块上锁 */
    private Object lock = new Object();

    List<Event> select() {
        return select(0);
    }

    List<Event> 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<Event> 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<InputSource> 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();
    }

}

接下篇：《Kafka网络模型基础-Reactor（下）》

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

相关推荐