背景
- verticle相当于1个执行模块，是vertx的部署单元。
- vertx可以部署多个verticle，且verticle之间可以互相通信。
- 因为vertx主要是个网络框架，所以在verticle中通常会启动server，如http或tcp。
Verticle部署源码分析
@Override
public void deployVerticle(String name, DeploymentOptions options, Handler<AsyncResult<String>> completionHandler) {
if (options.isHa() && haManager() != null && haManager().isEnabled()) {
haManager().deployVerticle(name, options, completionHandler);
} else {
deploymentManager.deployVerticle(name, options, completionHandler);
}
}
deploymentManager.deployVerticle代码如下：
public void deployVerticle(String identifier,
DeploymentOptions options,
Handler<AsyncResult<String>> completionHandler) {
ContextImpl callingContext = vertx.getOrCreateContext();
ClassLoader cl = getClassLoader(options, callingContext);
/**
* identifier为verticle类的全限定名
* callingContext为eventloop context
*/
doDeployVerticle(identifier, generateDeploymentID(), options, callingContext, callingContext, cl, completionHandler);
}
首先是创建context，ContextImpl callingContext = vertx.getOrCreateContext()，这块代码如下：
public ContextImpl getOrCreateContext() {
ContextImpl ctx = getContext();
if (ctx == null) {
// We are running embedded - Create a context
ctx = createEventLoopContext(null, null, new JsonObject(), Thread.currentThread().getContextClassLoader());
}
return ctx;
}
public ContextImpl getContext() {
ContextImpl context = (ContextImpl) context();
if (context != null && context.owner == this) {
return context;
}
return null;
}
public static Context context() {
Thread current = Thread.currentThread();
if (current instanceof VertxThread) {
return ((VertxThread) current).getContext();
}
return null;
}
public EventLoopContext createEventLoopContext(String deploymentID, WorkerPool workerPool, JsonObject config, ClassLoader tccl) {
return new EventLoopContext(this, internalBlockingPool, workerPool != null ? workerPool : this.workerPool, deploymentID, config, tccl);
}
最终会调到创建EventLoopContext的方法，详细见context部分分析，这里得到context后，接下来分析doDeployVerticle方法，如下：
private void doDeployVerticle(Iterator<VerticleFactory> iter,
Throwable prevErr,
String identifier,
String deploymentID,
DeploymentOptions options,
ContextImpl parentContext,
ContextImpl callingContext,
ClassLoader cl,
Handler<AsyncResult<String>> completionHandler) {
if (iter.hasNext()) {
VerticleFactory verticleFactory = iter.next();
Future<String> fut = Future.future();
if (verticleFactory.requiresResolve()) {
try {
verticleFactory.resolve(identifier, options, cl, fut);
} catch (Exception e) {
try {
fut.fail(e);
} catch (Exception ignore) {
// Too late
}
}
} else {
fut.complete(identifier);
}
fut.setHandler(ar -> {
Throwable err;
if (ar.succeeded()) {
/**
* resolvedName为verticle类全限定名
*/
String resolvedName = ar.result();
if (!resolvedName.equals(identifier)) {
deployVerticle(resolvedName, options, completionHandler);
return;
} else {
if (verticleFactory.blockingCreate()) {
vertx.<Verticle[]>executeBlocking(createFut -> {
try {
Verticle[] verticles = createVerticles(verticleFactory, identifier, options.getInstances(), cl);
createFut.complete(verticles);
} catch (Exception e) {
createFut.fail(e);
}
}, res -> {
if (res.succeeded()) {
doDeploy(identifier, deploymentID, options, parentContext, callingContext, completionHandler, cl, res.result());
} else {
// Try the next one
doDeployVerticle(iter, res.cause(), identifier, deploymentID, options, parentContext, callingContext, cl, completionHandler);
}
});
return;
} else {
try {
Verticle[] verticles = createVerticles(verticleFactory, identifier, options.getInstances(), cl);
doDeploy(identifier, deploymentID, options, parentContext, callingContext, completionHandler, cl, verticles);
return;
} catch (Exception e) {
err = e;
}
}
}
} else {
err = ar.cause();
}
// Try the next one
doDeployVerticle(iter, err, identifier, deploymentID, options, parentContext, callingContext, cl, completionHandler);
});
} else {
if (prevErr != null) {
// Report failure if there are no more factories to try otherwise try the next one
reportFailure(prevErr, callingContext, completionHandler);
} else {
// not handled or impossible ?
}
}
}
核心代码如下：
Verticle[] verticles = createVerticles(verticleFactory, identifier, options.getInstances(), cl);
doDeploy(identifier, deploymentID, options, parentContext, callingContext, completionHandler, cl, verticles);
createVerticles方法如下，创建verticle就是根据其类的全限定名创建对象：
private Verticle[] createVerticles(VerticleFactory verticleFactory, String identifier, int instances, ClassLoader cl) throws Exception {
Verticle[] verticles = new Verticle[instances];
for (int i = 0; i < instances; i++) {
verticles[i] = verticleFactory.createVerticle(identifier, cl);
if (verticles[i] == null) {
throw new NullPointerException("VerticleFactory::createVerticle returned null");
}
}
return verticles;
}
doDeploy方法如下：
private void doDeploy(String identifier, String deploymentID, DeploymentOptions options,
ContextImpl parentContext,
ContextImpl callingContext,
Handler<AsyncResult<String>> completionHandler,
ClassLoader tccl, Verticle... verticles) {
if (options.isMultiThreaded() && !options.isWorker()) {
throw new IllegalArgumentException("If multi-threaded then must be worker too");
}
JsonObject conf = options.getConfig() == null ? new JsonObject() : options.getConfig().copy(); // Copy it
String poolName = options.getWorkerPoolName();
Deployment parent = parentContext.getDeployment();
DeploymentImpl deployment = new DeploymentImpl(parent, deploymentID, identifier, options);
AtomicInteger deployCount = new AtomicInteger();
AtomicBoolean failureReported = new AtomicBoolean();
for (Verticle verticle : verticles) {
WorkerExecutorImpl workerExec = poolName != null ? vertx.createSharedWorkerExecutor(poolName, options.getWorkerPoolSize()) : null;
WorkerPool pool = workerExec != null ? workerExec.getPool() : null;
/**
* parentContext为部署verticle的context，这里又创建了新的context，用于启动verticle。
* 根据verticle类型，这里分为3种context，EventLoopContext、WorkerContext和MultiThreadedWorkerContext。
*/
ContextImpl context = options.isWorker() ? vertx.createWorkerContext(options.isMultiThreaded(), deploymentID, pool, conf, tccl) :
vertx.createEventLoopContext(deploymentID, pool, conf, tccl);
if (workerExec != null) {
context.addCloseHook(workerExec);
}
context.setDeployment(deployment);
deployment.addVerticle(new VerticleHolder(verticle, context));
/**
* 对于EventLoopContext，在context的eventloop中执行
* 对于WorkerContext，在worker线程顺序执行器中执行
* 对于MultiThreadedWorkerContext，在worker线程中执行
*/
context.runOnContext(v -> {
try {
verticle.init(vertx, context);
Future<Void> startFuture = Future.future();
verticle.start(startFuture);
startFuture.setHandler(ar -> {
if (ar.succeeded()) {
if (parent != null) {
parent.addChild(deployment);
deployment.child = true;
}
vertx.metricsSPI().verticleDeployed(verticle);
deployments.put(deploymentID, deployment);
if (deployCount.incrementAndGet() == verticles.length) {
reportSuccess(deploymentID, callingContext, completionHandler);
}
} else if (!failureReported.get()) {
reportFailure(ar.cause(), callingContext, completionHandler);
}
});
} catch (Throwable t) {
reportFailure(t, callingContext, completionHandler);
}
});
}
}
doDeploy核心代码主如下：
首先创建上下文：
ContextImpl context = options.isWorker() ? vertx.createWorkerContext(options.isMultiThreaded(), deploymentID, pool, conf, tccl) :
vertx.createEventLoopContext(deploymentID, pool, conf, tccl);
parentContext为部署verticle的context，这里又创建了新的context，用于启动verticle。
根据verticle类型，这里分为3种context，EventLoopContext、WorkerContext和MultiThreadedWorkerContext。
然后启动verticle，调用其init和start方法，但执行线程会因为verticle的不同模式而有所不同：
- 对于EventLoopContext，在context的eventloop中执行
- 对于WorkerContext，在worker线程顺序执行器中执行
- 对于MultiThreadedWorkerContext，在worker线程中执行
下面看下context.runOnContext方法，这里传入的回调是执行verticle的启动和初始化，除了这些，看还做了什么，以standard模式verticle为例，该context为EventLoopContext，runOnContext方法如下：
//EventLoopContext
public void executeAsync(Handler<Void> task) {
nettyEventLoop().execute(wrapTask(null, task, true, null));
}
nettyEventLoop()得到的是该context绑定的event loop，task回调参数即为verticle启动和初始化的回调，看看wrapTask都做了啥：
protected Runnable wrapTask(ContextTask cTask, Handler<Void> hTask, boolean checkThread, PoolMetrics metrics) {
Object metric = metrics != null ? metrics.submitted() : null;
return () -> {
Thread th = Thread.currentThread();
if (!(th instanceof VertxThread)) {
throw new IllegalStateException("Uh oh! Event loop context executing with wrong thread! Expected " + contextThread + " got " + th);
}
VertxThread current = (VertxThread) th;
if (THREAD_CHECKS && checkThread) {
if (contextThread == null) {
contextThread = current;
} else if (contextThread != current && !contextThread.isWorker()) {
throw new IllegalStateException("Uh oh! Event loop context executing with wrong thread! Expected " + contextThread + " got " + current);
}
}
if (metrics != null) {
metrics.begin(metric);
}
if (!DISABLE_TIMINGS) {
//记录开始时间
current.executeStart();
}
try {
//为当前线程current设置为当前上下文ContextImpl.this
setContext(current, ContextImpl.this);
if (cTask != null) {
cTask.run();
} else {
//执行启动和初始化verticle的回调
hTask.handle(null);
}
if (metrics != null) {
metrics.end(metric, true);
}
} catch (Throwable t) {
log.error("Unhandled exception", t);
Handler<Throwable> handler = this.exceptionHandler;
if (handler == null) {
handler = owner.exceptionHandler();
}
if (handler != null) {
handler.handle(t);
}
if (metrics != null) {
metrics.end(metric, false);
}
} finally {
// We don't unset the context after execution - this is done later when the context is closed via
// VertxThreadFactory
if (!DISABLE_TIMINGS) {
//清楚开始时间
current.executeEnd();
}
}
};
}
这段代码主要有4处核心调用：
- current.executeStart()：记录线程执行开始时间，为线程阻塞检查器提供线程开始时间信息。
//VertxThread
public final void executeStart() {
execStart = System.nanoTime();
}
- setContext(current, ContextImpl.this)：设置线程上下文
//ContextImpl
private static void setContext(VertxThread thread, ContextImpl context) {
thread.setContext(context);
if (!DISABLE_TCCL) {
if (context != null) {
context.setTCCL();
} else {
Thread.currentThread().setContextClassLoader(null);
}
}
}
//VertxThread
void setContext(ContextImpl context) {
this.context = context;
}
- hTask.handle(null)：执行verticle启动和初始化回调
- current.executeEnd()：清楚开始时间
//VertxThread
public final void executeEnd() {
execStart = 0;
}
网友评论