Vertx

背景

• 对于vertx框架，最基础的组件就是vertx，几乎所有功能都是通过vertx实现。
• vertx聚合了网络通信组件需要的eventloop和线程池。
• 所有verticle共享1个vertx实例。

vertx.png

VertxImpl核心成员

• DeploymentManager deploymentManager
• vertx部署verticle通过该成员实现。

this.deploymentManager = new DeploymentManager(this);

• BlockedThreadChecker checker
• 线程阻塞检查器，实现原理是执行一个定时任务，每次遍历所有VertxThread类型线程，如果线程执行时间超过了WarningExceptionTime，则记录日志。

checker = new BlockedThreadChecker(options.getBlockedThreadCheckInterval(), options.getWarningExceptionTime());

• ThreadFactory eventLoopThreadFactory
• eventloop线程创建工厂类，eventloop中的线程类型为VertxThread。

eventLoopThreadFactory = new VertxThreadFactory("vert.x-eventloop-thread-", checker, false, options.getMaxEventLoopExecuteTime());

• NioEventLoopGroup acceptorEventLoopGroup
• 请求接收eventloopGroup，负责accept连接，并把创建好的连接注册到eventLoopGroup中的某个eventloop上。这个group中只包含1个eventloop，即1个线程处理所有连接的accept请求。
• 线程类型为VertxThread。

ThreadFactory acceptorEventLoopThreadFactory = new VertxThreadFactory("vert.x-acceptor-thread-", checker, false, options.getMaxEventLoopExecuteTime());
acceptorEventLoopGroup = new NioEventLoopGroup(1, acceptorEventLoopThreadFactory);
acceptorEventLoopGroup.setIoRatio(100);

• NioEventLoopGroup eventLoopGroup
• 负责处理网络读写事件。这个group大小为通过vertx设置的eventloop大小，即eventLoopPoolSize。
• 线程类型为VertxThread。

eventLoopGroup = new NioEventLoopGroup(options.getEventLoopPoolSize(), eventLoopThreadFactory);
eventLoopGroup.setIoRatio(NETTY_IO_RATIO);

• WorkerPool workerPool
• worker线程池，用于执行耗时任务，在调用vertx.executeBlocking方法的Handler回调会在worker线程池中执行。
• 线程类型为VertxThread。

ExecutorService workerExec = Executors.newFixedThreadPool(options.getWorkerPoolSize(),
        new VertxThreadFactory("vert.x-worker-thread-", checker, true, options.getMaxWorkerExecuteTime()));

PoolMetrics workerPoolMetrics = isMetricsEnabled() ? metrics.createMetrics(workerExec, "worker", "vert.x-worker-thread", options.getWorkerPoolSize()) : null;

workerPool = new WorkerPool(workerExec, workerPoolMetrics);

• WorkerPool internalBlockingPool:
• 内部线程池，用于执行内部Action类型回调。
• 线程类型为VertxThread。

ExecutorService internalBlockingExec = Executors.newFixedThreadPool(options.getInternalBlockingPoolSize(),
        new VertxThreadFactory("vert.x-internal-blocking-", checker, true, options.getMaxWorkerExecuteTime()));

PoolMetrics internalBlockingPoolMetrics = isMetricsEnabled() ? metrics.createMetrics(internalBlockingExec, "worker", "vert.x-internal-blocking", options.getInternalBlockingPoolSize()) : null;

internalBlockingPool = new WorkerPool(internalBlockingExec, internalBlockingPoolMetrics);