###1.简单介绍
Quartz的功能十分强大,我这里只是简单记录我在项目中使用的过程,便于以后查缺补漏. Quartz的基本知识请看这里.由于我使用的版本为最新版(Version:2.2.2),因此和此篇文章的方法实现还是有小部分差异.
###2.需求说明
由于我需要实现一个定时任务,需要随意的更改定时时间或者取消定时任务.并且在程序崩溃重启后能够重新运行那些还未结束的定时任务. 自带的并不能完全满足需求.同时Quartz的jobName式的key->value异步实现方式配合来使用可谓事半功倍.
Quartz的持久化存储,还在研究....
本次的key值我暂时存储在redis中,每次新增一个定时任务将jobName 放在redis 中,完成定时任务移除redis中的jobName.系统重启前从redis中分别取出未完成的JobName 进行重新计算.
###3.功能实现
- 导入sbt文件
libraryDependencies += "org.quartz-scheduler" % "quartz" % "2.2.2"
2.QuartzManager工具类
package utils/** * Created by jiang on 16/1/18. */import models.basic._import org.quartz.DateBuilder._import org.quartz._import org.quartz.impl.StdSchedulerFactoryimport org.quartz.JobBuilder._import org.quartz.CronScheduleBuilder._import org.quartz.TriggerBuilder._import java.util.{HashMap, Date}import play.api.Loggerimport scala.collection.JavaConverters._import org.quartz.impl.matchers.GroupMatcher/** * Created by jiang on 16/1/18. */object QuartzManager { val scheduler = new StdSchedulerFactory().getScheduler() val jobGroupName = "jobs" val triggerGroupName = "triggers" // 在Global.onStart是提前初始化运行 def initialize() = { scheduler.clear() scheduler.start() } def getJobKey(jobName: String): JobKey = { val group = GroupMatcher.groupEquals[JobKey](jobGroupName) val keys = scheduler.getJobKeys(group).asScala.filter(_.getName == jobName) if (keys.nonEmpty) keys.head else null } def jobExists(jobName: String): Boolean = { getJobKey(jobName) != null }// 新增定时任务,只执行一次.同时将wid存入JobDataMap中.// stage 参数为不同阶段业务需求,存放在redis hash 不同的key关键字 def addJob(wid:Long,stage:String,time:Long,jobs:Job) = { var ret:Option[Int] = None val jobName,triggerName = wid+":"+stage val data = new JobDataMap() data.put("wid", wid) if(jobExists(jobName)) { Logger.info(s"$wid :后台已经有一个定时任务,请先取消再重新设置") ret = Option(0) }else{ val job: JobDetail = newJob(jobs.getClass) .withIdentity(jobName,jobGroupName) .usingJobData(data) .build() val trigger = newTrigger .withIdentity(triggerName,triggerGroupName) .startAt(new Date(time)) .forJob(jobName, jobGroupName) .build() scheduler.scheduleJob(job, trigger) // 将key 值存入redis 中 val key = "WorkTask:"+stage RedisUtils.getJedis.hset(key,wid.toString,time.toString) ret = Option(1) } // O-> 已存在,1->设置成功,none->未知错误 ret } def deleteJob(wid:Long,stage:String) ={ val jobName = wid+":"+stage scheduler.deleteJob(getJobKey(jobName)) // redis 中同样删除数据 val key = "WorkTask:"+stage RedisUtils.getJedis.hdel(key,wid.toString) } // 系统突然down 掉,重新恢复定时任务 def recoverJob() = { lazy val time = System.currentTimeMillis() // 当前时间,redis 中时间若小于当前时间则立即执行. val jedis = RedisUtils.getJedis //1.恢复准备中的定时任务 val readyWidKeyMap = jedis.hgetAll("WorkTask:ready").asScala if(readyWidKeyMap.nonEmpty){ readyWidKeyMap.foreach(p=>{ val wid:Long = p._1.toLong val recoverReadyTime:Long = if(p._2.toLong <= time ) time else p._2.toLong addJob(wid,"ready",recoverReadyTime,new readyJob()) }) } //2. 恢复进行中的定时任务 val ongoingWidKeyMap = jedis.hgetAll("WorkTask:ongoing").asScala if(ongoingWidKeyMap.nonEmpty){ ongoingWidKeyMap.foreach(p=>{ val wid:Long = p._1.toLong val recoverOnGoingTime:Long = if(p._2.toLong <= time ) time+5000 else p._2.toLong addJob(wid,"ongoing",recoverOnGoingTime,new ongoingJob()) }) } //3.恢复结束定时任务 val endWidKeyMap = jedis.hgetAll("WorkTask:end").asScala if(endWidKeyMap.nonEmpty){ endWidKeyMap.foreach(p=>{ val wid:Long = p._1.toLong val recoverEndTime:Long = if(p._2.toLong <= time ) time+10000 else p._2.toLong addJob(wid,"end",recoverEndTime,new endJob()) }) } }} 3.定时任务实现功能
class TaskJob extends Job{ val DateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); val d = new Date() val returnStr = DateFormat.format(d) override def execute(jobExecutionContext: JobExecutionContext): Unit ={ val wid = jobExecutionContext.getJobDetail.getJobDataMap.get("wid") println(returnStr +"★★★★★★★★★★★:"+wid) }} 4.测试
// 当前时间后的10执行val startTime = System.currentTimeMillis() + 10000QuartzManager.addJob(2,"ready",startTime,new TaskJob() )//10s 后,在控制台可以看到2016-01-19 11:01:53★★★★★★★★★★★:2
总结
我都写了啥啊?