Vert.X简介

Vert.X 是一个基于JVM的响应式应用/工具包（Reactive applications）。Vert.X是事件驱动，非阻塞纯异步IO : 极少线程资源处理大量并发请求，高并发系统的优选

官网上Vert.X呈现的3大特点：1. 节约资源/高并发处理优秀 2. 异步编程变得简单 3. 灵活，它不是一个框架，你可以只用它的某些模块之类

事件驱动

事件驱动 - 框架，软件设计模式/模型。与传统的函数调用区别：传统函数调用一定要有一个调用者并输入参数最后得到结果。事件驱动没有调用者，只关心响应（如何处理），它有一个事件收集器，谁产生了不是很重要，也不知道，收集到了这个事件就响应。事件驱动的核心是事件，关注的是响应（我关心的事件发生了如何处理）。经常使用在I/O框架里面，可以很好的实现I/O复用。

Reactor模式

在Web服务中，处理Web请求通常有两种体系结构，一个是基于线程的架构（Thread-based architecture）,一个是事件驱动模型（Event-driven architecture）

基于线程

缺点：

• 线程本身占用内存
• 创建销毁线程需要代价（线程池）
• 操作系统线程切换的开销
• 线程中处理I/O,等待过程造成CPU浪费

事件驱动

Reactor单线程

单线程：针对I/O的accept(),read(),write(),connect()是一个线程完成的，把非I/O业务逻辑从Reactor线程摘除，加速Reactor线程响应I/O请求

Reactor工作者线程池

与单线程的唯一区别是非I/O业务逻辑使用了一个工作线程池

Reactor 多线程

把Reactor拆分成mainReactor和subReactor。mainReactor(一个线程)负责监听server socket, 处理网络新连接，将建立的socketChannel指定注册给subReactor。subReactor维护自己的selector, 基于mainReactor 注册的socketChannel多路分离I/O读写事件，读写网络数据，通常使用多线程

• mainReactor对应Netty中配置的BossGroup线程组，主要负责接受客户端连接的建立。一般只暴露一个服务端口，BossGroup线程组一般一个线程工作即可
• subReactor对应Netty中配置的WorkerGroup线程组，BossGroup线程组接受并建立完客户端的连接后，将网络socket转交给WorkerGroup线程组，然后在WorkerGroup线程组内选择一个线程，进行I/O的处理。WorkerGroup线程组主要处理I/O，一般设置2*CPU核数个线程

package reactor.pattern;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;

/**
 * An Actor class which accepts the incoming connections
 */
public class Acceptor {

    private ServerSocketChannel serverSocketChannel;
    private Reactor[] reactors;

    private Selector selector;
    private int noOfReactors;

    public Acceptor(String host, int port, int noOfWorkerThreads) {
        try {
            selector = Selector.open();
            reactors = new Reactor[noOfWorkerThreads];
            this.noOfReactors = noOfWorkerThreads;
            for (int i = 0; i < noOfWorkerThreads; i++) {
                reactors[i] = new Reactor();
                Thread thread = new Thread(reactors[i]);
                thread.start();
            }

            serverSocketChannel = ServerSocketChannel.open();
            serverSocketChannel.bind(new InetSocketAddress(host, 9000));
            serverSocketChannel.configureBlocking(false);

            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);

        } catch (IOException e) {
            //handle exceptions
        }
    }

    public void start() throws IOException {

        int i = 0;

        while (true) {

            int readyChannels = selector.select();
            if (readyChannels == 0)
                continue;

            Set<SelectionKey> selectedKeys = selector.selectedKeys();

            Iterator<SelectionKey> keyIterator = selectedKeys.iterator();

            while (keyIterator.hasNext()) {

                SelectionKey key = keyIterator.next();

                if (key.isAcceptable()) {

                    ServerSocketChannel serverSocket = (ServerSocketChannel) key.channel();
                    SocketChannel socket = serverSocket.accept();
                    reactors[i % noOfReactors].addChannel(socket);
                    i++;

                }

                keyIterator.remove();
            }
        }
    }

    public static void main(String[] args) {
        Acceptor acceptor = new Acceptor("localhost",9000,4);
        try {
            acceptor.start();
        } catch (IOException e) {
           // e.printStackTrace();
        }
    }
}

package reactor.pattern;

import java.io.IOException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;

public class Reactor implements Runnable {

    private Queue queue = new ConcurrentLinkedQueue<SocketChannel>();

    private Selector selector;

    public Reactor() {
        try {
            this.selector = Selector.open();
        } catch (IOException e) {
            // e.printStackTrace();
        }
    }

    public void addChannel(SocketChannel socketChannel) {
        queue.add(socketChannel);
    }

    @Override
    public void run() {

        while (true) {

            try {
                SocketChannel socketChannel = (SocketChannel) queue.poll();
                if (socketChannel == null)
                    continue;

                socketChannel.configureBlocking(false);
                socketChannel.register(selector, SelectionKey.OP_READ);

                int readyChannels = selector.select();

                if (readyChannels == 0)
                    continue;

                Set<SelectionKey> selectedKeys = selector.selectedKeys();

                Iterator<SelectionKey> keyIterator = selectedKeys.iterator();

                while (keyIterator.hasNext()) {

                    SelectionKey key = keyIterator.next();

                    if (key.isReadable()) {
                        // Read the channel

                    }

                    keyIterator.remove();
                }
            } catch (IOException e) {
                //handle IOExceptions
            }
        }
    }
}

参考：

• https://medium.com/javarevisited/comparing-performance-and-developer-experience-of-migrating-a-spring-boot-back-end-to-vert-x-8f3281b4814#:~:text=Spring Boot had a throughput,five times greater than Vert.
• https://www.tibco.com/zh-hans/reference-center/what-is-event-driven-architecture
• https://juejin.cn/post/6983570550417424392
• https://zhuanlan.zhihu.com/p/95662364
• https://houbb.github.io/2017/11/16/netty-08-module-linux-05-reactor
• https://medium.com/@isururanawaka/implementation-of-reactor-pattern-using-java-selector-cda9fe1691eb