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

}


接下篇:《Kafka网络模型基础-Reactor(下)

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