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
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=MySqlDS org.quartz.jobStore.isClustered = true org.quartz.jobStore.clusterCheckinInterval = 5000 org.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;
@PostConstruct
public 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);
}
}
}
@PreDestroy
public 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: 0 Using 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 {
@Override
public 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
@Any
private Instance<Job> jobs;
@Override
public 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 too
private SimpleEjb simpleEjb;
@Override
public 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;
@Inject
private JobFactory cdiJobFactory;
@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();
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
}
@PreDestroy
public void stopJobs() {
// not changed
}
}
注意：在运行应用程序之前，将bean.xml文件添加到WEB-INF目录。
<?xml version="1.0" encoding="UTF-8"?>
<beans
xmlns="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中克隆了两次服务器配置，并为每个新副本分配了不同的端口偏移。
我需要做的其他更改是修改作业和触发器的创建。 由于所有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();
}
运行所有三个服务器实例后，我观察了结果。
工作执行
我观察到在所有三个节点上都执行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创建了运行配置：
然后我运行该应用程序，得到与JBoss 7相同的结果。
我发现WildFly似乎为持久EJB计时器提供了基于数据库的存储 ，但是我尚未对其进行调查。 也许是另一篇博客文章。
源代码
- 请在GitHub上找到此博客文章的源代码： https : //github.com/kolorobot/quartz-jee-demo