HOW-TO：带有MySQL的JEE应用程序中具有集群功能的Quartz Scheduler

Quartz Scheduler是Java世界中最流行的调度库之一。过去，我主要在Spring应用程序中使用Quartz。最近，我一直在研究将在云中部署的JBoss 7.1.1上运行的JEE 6应用程序中的调度。我考虑的一种选择是Quartz Scheduler，因为它提供了与数据库的集群。在本文中，我将展示在JEE应用程序中配置Quartz并在JBoss 7.1.1或WildFly 8.0.0上运行它，使用MySQL作为作业存储并利用CDI在作业中使用依赖注入是多么容易。所有这些都将在IntelliJ中完成。让我们开始吧。

创建Maven项目

我使用org.codehaus.mojo.archetypes:webapp-javaee6原型来引导应用程序，然后我对pom.xml进行了一些修改。我还添加了slf4J依赖项，因此生成的pom.xml如下所示：

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>pl.codeleak</groupId><artifactId>quartz-jee-demo</artifactId><version>1.0</version><packaging>war</packaging><name>quartz-jee-demo</name><properties><endorsed.dir>${project.build.directory}/endorsed</endorsed.dir><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding></properties><dependencies><dependency><groupId>javax</groupId><artifactId>javaee-api</artifactId><version>6.0</version><scope>provided</scope></dependency><!-- Logging --><dependency><groupId>org.slf4j</groupId><artifactId>slf4j-api</artifactId><version>1.7.7</version></dependency><dependency><groupId>org.slf4j</groupId><artifactId>slf4j-jdk14</artifactId><version>1.7.7</version></dependency></dependencies><build><plugins><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-compiler-plugin</artifactId><version>2.3.2</version><configuration><source>1.7</source><target>1.7</target><compilerArguments><endorseddirs>${endorsed.dir}</endorseddirs></compilerArguments></configuration></plugin><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-war-plugin</artifactId><version>2.1.1</version><configuration><failOnMissingWebXml>false</failOnMissingWebXml></configuration></plugin><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-dependency-plugin</artifactId><version>2.1</version><executions><execution><phase>validate</phase><goals><goal>copy</goal></goals><configuration><outputDirectory>${endorsed.dir}</outputDirectory><silent>true</silent><artifactItems><artifactItem><groupId>javax</groupId><artifactId>javaee-endorsed-api</artifactId><version>6.0</version><type>jar</type></artifactItem></artifactItems></configuration></execution></executions></plugin></plugins></build></project>

接下来是将项目导入到IDE。在我的情况下，这是IntelliJ，并使用JBoss 7.1.1创建运行配置。

值得注意的是，在运行配置中的VM Options中，我添加了两个变量：

-Djboss.server.default.config=standalone-custom.xml
-Djboss.socket.binding.port-offset=100

带有群集的石英计划程序img1

standalone-custom.xml是标准standalone.xml的副本，因为需要修改配置（请参见下文）。

配置JBoss服务器

在我的演示应用程序中，我想将MySQL数据库与Quartz一起使用，因此需要将MySQL数据源添加到我的配置中。这可以通过两个步骤快速完成。

添加驱动程序模块

我创建了一个文件夹JBOSS_HOME/modules/com/mysql/main 。在这个文件夹中，我添加了两个文件： module.xml和mysql-connector-java-5.1.23.jar 。模块文件如下所示：

<?xml version="1.0" encoding="UTF-8"?>  
<module xmlns="urn:jboss:module:1.0" name="com.mysql">  <resources>  <resource-root path="mysql-connector-java-5.1.23.jar"/>  </resources>  <dependencies>  <module name="javax.api"/>  </dependencies>  
</module>

配置数据源

在datasources子系统的standalone-custom.xml文件中，我添加了一个新的数据源：

<datasource jta="false" jndi-name="java:jboss/datasources/MySqlDS" pool-name="MySqlDS" enabled="true" use-java-context="true"><connection-url>jdbc:mysql://localhost:3306/javaee</connection-url><driver>com.mysql</driver><security><user-name>jeeuser</user-name><password>pass</password></security>
</datasource>

和驱动程序：

<drivers><driver name="com.mysql" module="com.mysql"/>
</drivers>

注意：就本演示而言，数据源不是由JTA管理的，以简化配置。

使用集群配置Quartz

我使用了官方教程通过集群配置Quarts： http : //quartz-scheduler.org/documentation/quartz-2.2.x/configuration/ConfigJDBCJobStoreClustering

将Quartz依赖项添加到`pom.xml`

<dependency><groupId>org.quartz-scheduler</groupId><artifactId>quartz</artifactId><version>2.2.1</version>
</dependency>
<dependency><groupId>org.quartz-scheduler</groupId><artifactId>quartz-jobs</artifactId><version>2.2.1</version>
</dependency>

将`quartz.properties`添加到`src/main/resources`

#============================================================================
# Configure Main Scheduler Properties  
#============================================================================org.quartz.scheduler.instanceName = MyScheduler
org.quartz.scheduler.instanceId = AUTO#============================================================================
# Configure ThreadPool  
#============================================================================org.quartz.threadPool.class = org.quartz.simpl.SimpleThreadPool
org.quartz.threadPool.threadCount = 1#============================================================================
# Configure JobStore  
#============================================================================org.quartz.jobStore.class = org.quartz.impl.jdbcjobstore.JobStoreTX
org.quartz.jobStore.driverDelegateClass=org.quartz.impl.jdbcjobstore.StdJDBCDelegate
org.quartz.jobStore.useProperties = false
org.quartz.jobStore.dataSource=MySqlDSorg.quartz.jobStore.isClustered = true
org.quartz.jobStore.clusterCheckinInterval = 5000org.quartz.dataSource.MySqlDS.jndiURL=java:jboss/datasources/MySqlDS

创建供Quartz使用的MySQL表

可以在Quartz发行版中找到模式文件： quartz-2.2.1\docs\dbTables 。

演示代码

有了适当的配置后，我想检查Quartz是否工作，因此我创建了一个没有作业和触发器的调度程序。

package pl.codeleak.quartzdemo;import org.quartz.JobKey;
import org.quartz.Scheduler;
import org.quartz.SchedulerException;
import org.quartz.TriggerKey;
import org.quartz.impl.StdSchedulerFactory;
import org.quartz.impl.matchers.GroupMatcher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.ejb.Singleton;
import javax.ejb.Startup;@Startup
@Singleton
public class SchedulerBean {private Logger LOG = LoggerFactory.getLogger(SchedulerBean.class);private Scheduler scheduler;@PostConstructpublic void scheduleJobs() {try {scheduler = new StdSchedulerFactory().getScheduler();            scheduler.start();printJobsAndTriggers(scheduler);} catch (SchedulerException e) {LOG.error("Error while creating scheduler", e);}}private void printJobsAndTriggers(Scheduler scheduler) throws SchedulerException {LOG.info("Quartz Scheduler: {}", scheduler.getSchedulerName());for(String group: scheduler.getJobGroupNames()) {for(JobKey jobKey : scheduler.getJobKeys(GroupMatcher.<JobKey>groupEquals(group))) {LOG.info("Found job identified by {}", jobKey);}}for(String group: scheduler.getTriggerGroupNames()) {for(TriggerKey triggerKey : scheduler.getTriggerKeys(GroupMatcher.<TriggerKey>groupEquals(group))) {LOG.info("Found trigger identified by {}", triggerKey);}}}@PreDestroypublic void stopJobs() {if (scheduler != null) {try {scheduler.shutdown(false);} catch (SchedulerException e) {LOG.error("Error while closing scheduler", e);}}}
}

运行应用程序时，您应该能够从Quartz中看到一些调试信息：

Scheduler class: 'org.quartz.core.QuartzScheduler' - running locally.NOT STARTED.Currently in standby mode.Number of jobs executed: 0Using thread pool 'org.quartz.simpl.SimpleThreadPool' - with 1 threads.Using job-store 'org.quartz.impl.jdbcjobstore.JobStoreTX' - which supports persistence. and is clustered.

让Quartz利用CDI

在Quartz中，作业必须实现org.quartz.Job接口。

package pl.codeleak.quartzdemo;import org.quartz.Job;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;public class SimpleJob implements Job {@Overridepublic void execute(JobExecutionContext context) throws JobExecutionException {// do something}
}

然后使用JobBuilder创建一个Job：

JobKey job1Key = JobKey.jobKey("job1", "my-jobs");
JobDetail job1 = JobBuilder.newJob(SimpleJob.class).withIdentity(job1Key).build();

在我的示例中，我需要将EJB注入到我的作业中，以便重新使用现有的应用程序逻辑。因此，实际上，我需要注入一个EJB参考。 Quartz如何做到这一点？简单。 Quartz Scheduler有一个提供JobFactory的方法，该方法将负责创建Job实例。

package pl.codeleak.quartzdemo;import org.quartz.Job;
import org.quartz.JobDetail;
import org.quartz.Scheduler;
import org.quartz.SchedulerException;
import org.quartz.spi.JobFactory;
import org.quartz.spi.TriggerFiredBundle;import javax.enterprise.inject.Any;
import javax.enterprise.inject.Instance;
import javax.inject.Inject;
import javax.inject.Named;public class CdiJobFactory implements JobFactory {@Inject@Anyprivate Instance<Job> jobs;@Overridepublic Job newJob(TriggerFiredBundle triggerFiredBundle, Scheduler scheduler) throws SchedulerException {final JobDetail jobDetail = triggerFiredBundle.getJobDetail();final Class<? extends Job> jobClass = jobDetail.getJobClass();for (Job job : jobs) {if (job.getClass().isAssignableFrom(jobClass)) {return job;}}throw new RuntimeException("Cannot create a Job of type " + jobClass);}
}

到目前为止，所有作业都可以使用依赖项注入和注入其他依赖项，包括EJB。

package pl.codeleak.quartzdemo.ejb;import org.slf4j.Logger;
import org.slf4j.LoggerFactory;import javax.ejb.Stateless;@Stateless
public class SimpleEjb {private static final Logger LOG = LoggerFactory.getLogger(SimpleEjb.class);public void doSomething() {LOG.info("Inside an EJB");}
}package pl.codeleak.quartzdemo;import org.quartz.Job;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import pl.codeleak.quartzdemo.ejb.SimpleEjb;import javax.ejb.EJB;
import javax.inject.Named;public class SimpleJob implements Job {@EJB // @Inject will work tooprivate SimpleEjb simpleEjb;@Overridepublic void execute(JobExecutionContext context) throws JobExecutionException {simpleEjb.doSomething();}
}

最后一步是修改SchedulerBean：

package pl.codeleak.quartzdemo;import org.quartz.*;
import org.quartz.impl.StdSchedulerFactory;
import org.quartz.impl.matchers.GroupMatcher;
import org.quartz.spi.JobFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.ejb.Singleton;
import javax.ejb.Startup;
import javax.inject.Inject;@Startup
@Singleton
public class SchedulerBean {private Logger LOG = LoggerFactory.getLogger(SchedulerBean.class);private Scheduler scheduler;@Injectprivate JobFactory cdiJobFactory;@PostConstructpublic void scheduleJobs() {try {scheduler = new StdSchedulerFactory().getScheduler();scheduler.setJobFactory(cdiJobFactory);JobKey job1Key = JobKey.jobKey("job1", "my-jobs");JobDetail job1 = JobBuilder.newJob(SimpleJob.class).withIdentity(job1Key).build();TriggerKey tk1 = TriggerKey.triggerKey("trigger1", "my-jobs");Trigger trigger1 = TriggerBuilder.newTrigger().withIdentity(tk1).startNow().withSchedule(SimpleScheduleBuilder.repeatSecondlyForever(10)).build();scheduler.scheduleJob(job1, trigger1);scheduler.start();printJobsAndTriggers(scheduler);} catch (SchedulerException e) {LOG.error("Error while creating scheduler", e);}}private void printJobsAndTriggers(Scheduler scheduler) throws SchedulerException {// not changed}@PreDestroypublic void stopJobs() {// not changed}
}

注意：在运行应用程序之前，将bean.xml文件添加到WEB-INF目录。

<?xml version="1.0" encoding="UTF-8"?>
<beansxmlns="http://xmlns.jcp.org/xml/ns/javaee"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://xmlns.jcp.org/xml/ns/javaee http://xmlns.jcp.org/xml/ns/javaee/beans_1_1.xsd"bean-discovery-mode="all"></beans>

现在，您可以启动服务器并观察结果。首先，创建作业和触发器：

12:08:19,592 INFO   (MSC service thread 1-3) Quartz Scheduler: MyScheduler
12:08:19,612 INFO   (MSC service thread 1-3) Found job identified by my-jobs.job1
12:08:19,616 INFO   (MSC service thread 1-3) Found trigger identified by m

我们的工作正在运行（大约每10秒运行一次）：

12:08:29,148 INFO   (MyScheduler_Worker-1) Inside an EJB
12:08:39,165 INFO   (MyScheduler_Worker-1) Inside an EJB

还要查看Quartz表内部，您将看到其中已填充了数据。

测试应用

我要检查的最后一件事是如何在多个实例中触发作业。为了进行测试，我只是在IntelliJ中克隆了两次服务器配置，并为每个新副本分配了不同的端口偏移。

带有群集的石英计划程序img2
我需要做的其他更改是修改作业和触发器的创建。由于所有Quartz对象都存储在数据库中，因此创建相同的作业和触发器（使用相同的键）将引发异常：

Error while creating scheduler: org.quartz.ObjectAlreadyExistsException: Unable to store Job : 'my-jobs.job1', because one already exists with this identification.

我需要更改代码，以确保如果作业/触发器存在，我将对其进行更新。此测试的scheduleJobs方法的最终代码为同一作业注册了三个触发器。

@PostConstruct
public void scheduleJobs() {try {scheduler = new StdSchedulerFactory().getScheduler();scheduler.setJobFactory(cdiJobFactory);JobKey job1Key = JobKey.jobKey("job1", "my-jobs");JobDetail job1 = JobBuilder.newJob(SimpleJob.class).withIdentity(job1Key).build();TriggerKey tk1 = TriggerKey.triggerKey("trigger1", "my-jobs");Trigger trigger1 = TriggerBuilder.newTrigger().withIdentity(tk1).startNow().withSchedule(SimpleScheduleBuilder.repeatSecondlyForever(10)).build();TriggerKey tk2 = TriggerKey.triggerKey("trigger2", "my-jobs");Trigger trigger2 = TriggerBuilder.newTrigger().withIdentity(tk2).startNow().withSchedule(SimpleScheduleBuilder.repeatSecondlyForever(10)).build();TriggerKey tk3 = TriggerKey.triggerKey("trigger3", "my-jobs");Trigger trigger3 = TriggerBuilder.newTrigger().withIdentity(tk3).startNow().withSchedule(SimpleScheduleBuilder.repeatSecondlyForever(10)).build();scheduler.scheduleJob(job1, newHashSet(trigger1, trigger2, trigger3), true);scheduler.start();printJobsAndTriggers(scheduler);} catch (SchedulerException e) {LOG.error("Error while creating scheduler", e);}
}

除上述内容外，我还添加了在SimpleJob中记录JobExecutionContext的信息，因此我可以更好地分析结果。

@Override
public void execute(JobExecutionContext context) throws JobExecutionException {try {LOG.info("Instance: {}, Trigger: {}, Fired at: {}",context.getScheduler().getSchedulerInstanceId(),context.getTrigger().getKey(),sdf.format(context.getFireTime()));} catch (SchedulerException e) {}simpleEjb.doSomething();
}

运行所有三个服务器实例后，我观察了结果。

带有群集的石英计划程序-img3

工作执行

我观察到在所有三个节点上都执行trigger2，并且在三个节点上执行了trigger2，如下所示：

Instance: kolorobot1399805959393 (instance1), Trigger: my-jobs.trigger2, Fired at: 13:00:09
Instance: kolorobot1399805989333 (instance3), Trigger: my-jobs.trigger2, Fired at: 13:00:19
Instance: kolorobot1399805963359 (instance2), Trigger: my-jobs.trigger2, Fired at: 13:00:29
Instance: kolorobot1399805959393 (instance1), Trigger: my-jobs.trigger2, Fired at: 13:00:39
Instance: kolorobot1399805959393 (instance1), Trigger: my-jobs.trigger2, Fired at: 13:00:59

对于其他触发器类似。

复苏

断开kolorobot1399805989333（instance3）的连接后，一段时间后，我在日志中看到以下内容：

ClusterManager: detected 1 failed or restarted instances.
ClusterManager: Scanning for instance "kolorobot1399805989333"'s failed in-progress jobs.

然后我断开了kolorobot1399805963359（instance2）的连接，这也是我在日志中看到的内容：

ClusterManager: detected 1 failed or restarted instances.
ClusterManager: Scanning for instance "kolorobot1399805963359"'s failed in-progress jobs.
ClusterManager: ......Freed 1 acquired trigger(s).

到目前为止，由kolorobot1399805959393（instance1）执行的所有触发器

在Wildfly 8上运行

无需任何更改，我就可以在WildFly 8.0.0上部署相同的应用程序。与JBoss 7.1.1相似，我添加了MySQL模块（WildFly 8上modules文件夹的位置不同– modules/system/layers/base/com/mysql/main 。数据源和驱动程序的定义与上图完全相同。我为WildFly 8创建了运行配置：

带有群集的石英计划程序-img4
然后我运行该应用程序，得到与JBoss 7相同的结果。