ElasticJob有一套自己的事件发布机制，核心部件使用的Guava的EventBus，如果不是很了解，出门左转看我的关于Guava的EventBus文章；
1，事件配置接口-JobEventConfiguration
public interface JobEventConfiguration extends JobEventIdentity {
JobEventListener createJobEventListener() throws JobEventListenerConfigurationException;
}
- 该接口非常简单，只是为了获取JobEventListener；
- JobEventListener的本质就是观察者，它的实现类将会使用EventBus的register方法注册到事件总线中；
2，事件的观察者-JobEventRdbListener
先来看看这个接口中定义了什么；
public interface JobEventListener extends JobEventIdentity {
@Subscribe
@AllowConcurrentEvents
void listen(JobExecutionEvent jobExecutionEvent);
@Subscribe
@AllowConcurrentEvents
void listen(JobStatusTraceEvent jobStatusTraceEvent);
}
- 接口中定义了两个方法，也就意味着ElasticJob只会对两个事件作出反应，这个接口没有拓展性，已经固定死只能处理两种事件，但是没有关系，ElasticJob只关心这两个事件，如果后期有更多的事件需要处理，只能在接口继续增加方法；
- 这两个方法都使用了 @Subscribe 和 @AllowConcurrentEvents 注解，也就意味着，这两个方法可以在多线程中并发执行，需要注意的一点是，该接口的实现类上最好不要再写@Subscribe注解，否则，因为实现类的方法上没有 @AllowConcurrentEvents注解，而实现类上的注解又是优先被解读，就会导致解读不到父类方法上的注解，从而导致方法失去了@AllowConcurrentEvents注解，也就是不能在多线程中同时调用，多线程需要排队获取同步锁；
再来看看该接口的实现类：
public final class JobEventRdbListener extends JobEventRdbIdentity implements JobEventListener {
private final JobEventRdbStorage repository;
public JobEventRdbListener(final DataSource dataSource) throws SQLException {
repository = new JobEventRdbStorage(dataSource);
}
@Override
public void listen(final JobExecutionEvent executionEvent) {
repository.addJobExecutionEvent(executionEvent);
}
@Override
public void listen(final JobStatusTraceEvent jobStatusTraceEvent) {
repository.addJobStatusTraceEvent(jobStatusTraceEvent);
}
}
- 可以看到该实现类复写的方法上没有使用任何注解；
- 该事件监听器主要用来持久化数据，使用到了JobEventRdbStorage类，该类在初始化时，会创建表，具体逻辑可自行看代码深入了解；
3，事件总线-JobEventBus
有了事件监听器（JobEventListener），并且有了能够生产事件监听器的工厂（JobEventConfiguration），那么下一步就是将事件监听器注册到事件总线中了，先来看一下ElasticJob的事件总线；
public final class JobEventBus {
private final JobEventConfiguration jobEventConfig;
private final ExecutorServiceObject executorServiceObject;
private final EventBus eventBus;
private boolean isRegistered;
public JobEventBus() {
jobEventConfig = null;
executorServiceObject = null;
eventBus = null;
}
public JobEventBus(final JobEventConfiguration jobEventConfig) {
this.jobEventConfig = jobEventConfig;
executorServiceObject = new ExecutorServiceObject("job-event", Runtime.getRuntime().availableProcessors() * 2);
eventBus = new AsyncEventBus(executorServiceObject.createExecutorService());
register();
}
private void register() {
try {
eventBus.register(jobEventConfig.createJobEventListener());
isRegistered = true;
} catch (final JobEventListenerConfigurationException ex) {
log.error("Elastic job: create JobEventListener failure, error is: ", ex);
}
}
public void post(final JobEvent event) {
if (isRegistered && !executorServiceObject.isShutdown()) {
eventBus.post(event);
}
}
}
解读：
- 首先来看一下空构造函数，所有值都为空，之所以这么写，是因为所有的事件的发布代码，都已经写到代码中，不可分离，此时又没有配置事件总线需要的数据，那么唯一的方法就是不发布事件，所以上面的post方法中，使用了isRegistered，当没有配置事件总线时，不发布任何事件；
- 即使不是使用空构造函数，如果在构建事件总线时报错，那么也一并认为事件总线不可用，捕获异常，isRegistered还是false，事件还是不发布；
- 这里的ExecutorServiceObject是获取线程池的工具类，默认线程数量为cpu核心的两倍；
- 当成功构建事件总线时，将事件监听器注册到EventBus中，并设置isRegistered为true，表示已注册，事件可用；
网友评论