一、定时任务与消息队列
- 本质区别
定时任务：时间驱动
消息队列：事件驱动 - 相同点
1.异步处理
比如注册、下单事件
2.应用解耦
定时任务作业还是MQ都可以作为两个应用之间的⻮轮实现应用解耦
3.流量削峰
二、单体任务调度（任务调度框架Quartz）
- 引入依赖
<!--任务调度框架quartz-->
<!-- https://mvnrepository.com/artifact/org.quartz-scheduler/quartz --
>
<dependency>
<groupId>org.quartz-scheduler</groupId>
<artifactId>quartz</artifactId>
<version>2.3.2</version>
</dependency>
- 定时任务作业主调度程序
package quartz;
import org.quartz.*;
import org.quartz.impl.StdSchedulerFactory;
public class QuartzMain {
// 创建作业任务调度器（类似于公交调度站）
public static Scheduler createScheduler() throws
SchedulerException {
SchedulerFactory schedulerFactory = new StdSchedulerFactory();
Scheduler scheduler = schedulerFactory.getScheduler();
return scheduler;
}
// 创建一个作业任务（类似于一辆公交⻋）
public static JobDetail createJob() {
JobBuilder jobBuilder = JobBuilder.newJob(DemoJob.class);
jobBuilder.withIdentity("jobName","myJob");
JobDetail jobDetail = jobBuilder.build();
return jobDetail;
}
/**
* 创建作业任务时间触发器（类似于公交⻋出⻋时间表）
* cron表达式由七个位置组成，空格分隔
* 1、Seconds（秒） 0~59
* 2、Minutes（分） 0~59
* 3、Hours（小时） 0~23
* 4、Day of Month（天）1~31,注意有的月份不足31天
* 5、Month（月） 0~11,或者
JAN,FEB,MAR,APR,MAY,JUN,JUL,AUG,SEP,OCT,NOV,DEC
* 6、Day of Week(周) 1~7,1=SUN或者 SUN,MON,TUE,WEB,THU,FRI,SAT
定义一个job，需实现Job接口
* 7、Year（年）1970~2099 可选项
*示例：
* 0 0 11 * * ? 每天的11点触发执行一次
* 0 30 10 1 * ? 每月1号上午10点半触发执行一次
*/
public static Trigger createTrigger() {
// 创建时间触发器，按日历调度
CronTrigger trigger = TriggerBuilder.newTrigger()
.withIdentity("triggerName","myTrigger")
.startNow()
.withSchedule(CronScheduleBuilder.cronSchedule("0/2 *
* * * ?"))
.build();
// 创建触发器，按简单间隔调度
/*SimpleTrigger trigger1 = TriggerBuilder.newTrigger()
.withIdentity("triggerName","myTrigger")
.startNow()
.withSchedule(SimpleScheduleBuilder
.simpleSchedule()
.withIntervalInSeconds(3)
.repeatForever())
.build();*/
return trigger;
}
// 定时任务作业主调度程序
public static void main(String[] args) throws SchedulerException {
// 创建一个作业任务调度器（类似于公交调度站）
Scheduler scheduler = QuartzMain.createScheduler();
// 创建一个作业任务（类似于一辆公交⻋）
JobDetail job = QuartzMain.createJob();
// 创建一个作业任务时间触发器（类似于公交⻋出⻋时间表）
Trigger trigger = QuartzMain.createTrigger();
// 使用调度器按照时间触发器执行这个作业任务
scheduler.scheduleJob(job,trigger);
scheduler.start();
}
}
- 定义一个job，需要实现job接口
package quartz;
import org.quartz.Job;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
public class DemoJob implements Job {
public void execute(JobExecutionContext jobExecutionContext)
throws JobExecutionException {
System.out.println("我是一个定时任务逻辑");
}
}
三、分布式调度框架Elastic-Job
当当网开源的分布式调度解决方案，基于Quartz二次开发
-
两个相互独立的子项目
Elastic-Job-Lite
轻量级无中心化解决方案，使用Jar包的形式提供分布式任务的协调服务Elastic-Job-Cloud组成
需要结合Mesos以及Docker在云环境下使用
主要功能介绍
分布式调度协调
在分布式环境中，任务能够按指定的调度策略执行，并且能够避免同一任务多实例重复执行丰富的调度策略 基于成熟的定时任务作业框架Quartz cron表达式执行定时任务弹性
扩容缩容 当集群中增加某一个实例，它应当也能够被选举并执行任务；当集群减少一个实例时，它所执行的任务能被转移到别的实例来执行。
失效转移 某实例在任务执行失败后，会被转移到其他实例执行
错过执行作业重触发 若因某种原因导致作业错过执行，自动记录错过执行的作业，并在上次作业完成后自动触发。
支持并行调度 支持任务分片，任务分片是指将一个任务分为多个小任务项在多个实例同时执行。
作业分片一致性 当任务被分片后，保证同一分片在分布式环境中仅一个执行实例。
Elastic-Job-Lite应用
jar包（API） + 安装zk软件（3.4.6版本以上/本质功能：存储+通知）
-
安装Zookeeper（此处单例配置）
1）我们使用3.4.10版本，在linux平台解压下载的zookeeper-3.4.10.tar.gz
2）进入conf目录，cp zoo_sample.cfg zoo.cfg
- 进入bin目录，启动zk服务
启动 ./zkServer.sh start
停止 ./zkServer.sh stop
查看状态 ./zkServer.sh status
- 进入bin目录，启动zk服务
引入Jar包
<!-- https://mvnrepository.com/artifact/com.dangdang/elastic-job-lite-core
-->
<dependency>
<groupId>com.dangdang</groupId>
<artifactId>elastic-job-lite-core</artifactId>
<version>2.1.5</version>
</dependency>
- 定时任务实例
需求：:每隔两秒钟执行一次定时任务（resume表中未归档的数据归档resume_bak表中，每次归档1条记录）
1）resume_bak和resume表结构完全一样
2）resume表中数据归档之后不删除，只将state置为"已归档"
数据表结构
-- ----------------------------
-- Table structure for resume
-- ----------------------------
DROP TABLE IF EXISTS `resume`;
CREATE TABLE `resume` (
`id` bigint(20) NOT NULL AUTO_INCREMENT,
`name` varchar(255) DEFAULT NULL,
`sex` varchar(255) DEFAULT NULL,
`phone` varchar(255) DEFAULT NULL,
`address` varchar(255) DEFAULT NULL,
`education` varchar(255) DEFAULT NULL,
`state` varchar(255) DEFAULT NULL,
PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=1001 DEFAULT CHARSET=utf8;
SET FOREIGN_KEY_CHECKS = 1;
程序开发:
- 定时任务类
package elasticjob;
import com.dangdang.ddframe.job.api.ShardingContext;
import com.dangdang.ddframe.job.api.simple.SimpleJob;
import util.JdbcUtil;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.List;
import java.util.Map;
public class BackupJob implements SimpleJob {
// 定时任务每执行一次都会执行如下的逻辑
@Override
public void execute(ShardingContext shardingContext) {
/*
从resume数据表查找1条未归档的数据，将其归档到resume_bak
表，并更新状态为已归档（不删除原数据）
*/
// 查询出一条数据
String selectSql = "select * from resume where state='未归档' limit 1";
List<Map<String, Object>> list = JdbcUtil.executeQuery(selectSql);
if(list == null || list.size() == 0) {
return;
}
Map<String, Object> stringObjectMap = list.get(0);
long id = (long) stringObjectMap.get("id");
String name = (String) stringObjectMap.get("name");
String education = (String) stringObjectMap.get("education");
// 打印出这条记录
System.out.println("======>>>id：" + id + " name：" + name + " education：" + education);
// 更改状态
String updateSql = "update resume set state='已归档' where id=?";
JdbcUtil.executeUpdate(updateSql,id);
// 归档这条记录
String insertSql = "insert into resume_bak select * from resume where id=?";
JdbcUtil.executeUpdate(insertSql,id);
}
}
- 主类
package elasticjob;
import com.dangdang.ddframe.job.config.JobCoreConfiguration;
import
com.dangdang.ddframe.job.config.simple.SimpleJobConfiguration;
import com.dangdang.ddframe.job.lite.api.JobScheduler;
import com.dangdang.ddframe.job.lite.config.LiteJobConfiguration;
import
com.dangdang.ddframe.job.reg.base.CoordinatorRegistryCenter;
import
com.dangdang.ddframe.job.reg.zookeeper.ZookeeperConfiguration;
import
com.dangdang.ddframe.job.reg.zookeeper.ZookeeperRegistryCenter;
public class ElasticJobMain {
public static void main(String[] args) {
// 配置注册中心zookeeper，zookeeper协调调度，不能让任务重复执行，通过命名空间分类管理任务，对应到zookeeper的目录
ZookeeperConfiguration zookeeperConfiguration = new ZookeeperConfiguration("localhost:2181","data-archive-job");
CoordinatorRegistryCenter coordinatorRegistryCenter = new ZookeeperRegistryCenter(zookeeperConfiguration);
coordinatorRegistryCenter.init();
// 配置任务
JobCoreConfiguration jobCoreConfiguration = JobCoreConfiguration.newBuilder("archive-job","*/2 * * * * ?",1).build();
SimpleJobConfiguration simpleJobConfiguration = new SimpleJobConfiguration(jobCoreConfiguration,BackupJob.class.getName());
// 启动任务
new JobScheduler(coordinatorRegistryCenter, LiteJobConfiguration.newBuilder(simpleJobConfiguration).build()).init();
}
}
- JdbcUtil工具类
package util;
import java.sql.*;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
public class JdbcUtil {
//url
private static String url = "jdbc:mysql://localhost:3306/job?
characterEncoding=utf8&useSSL=false";
//user
private static String user = "root";
//password
private static String password = "123456";
//驱动程序类
private static String driver = "com.mysql.jdbc.Driver";
static {
try {
Class.forName(driver);
} catch (ClassNotFoundException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
public static Connection getConnection() {
try {
return DriverManager.getConnection(url, user, password);
} catch (SQLException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
return null;
}
public static void close(ResultSet rs, PreparedStatement ps, Connection con) {
if (rs != null) {
try {
rs.close();
} catch (SQLException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} finally {
if (ps != null) {
try {
ps.close();
} catch (SQLException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} finally {
if (con != null) {
try {
con.close();
} catch (SQLException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
}
}
}
}
public static void executeUpdate(String sql,Object...obj) {
Connection con = getConnection();
PreparedStatement ps = null;
try {
ps = con.prepareStatement(sql);
for (int i = 0; i < obj.length; i++) {
ps.setObject(i + 1, obj[i]);
}
ps.executeUpdate();
} catch (SQLException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} finally {
close(null, ps, con);
}
}
public static List<Map<String,Object>> executeQuery(String sql, Object...obj) {
Connection con = getConnection();
ResultSet rs = null;
PreparedStatement ps = null;
try {
ps = con.prepareStatement(sql);
for (int i = 0; i < obj.length; i++) {
ps.setObject(i + 1, obj[i]);
}
rs = ps.executeQuery();
List<Map<String, Object>> list = new ArrayList<>();
int count = rs.getMetaData().getColumnCount();
while (rs.next()) {
Map<String, Object> map = new HashMap<String, Object>();
for (int i = 0; i < count; i++) {
Object ob = rs.getObject(i + 1);
String key = rs.getMetaData().getColumnName(i + 1);
map.put(key, ob);
}
list.add(map);
}
return list;
} catch (SQLException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} finally {
close(rs, ps, con);
}
return null;
}
}
- 测试
4.1 先启动一个进程，然后再启动一个进程（两个进程模拟分布式环境下，通一个定时任务部署了两份在工作）
4.2 两个进程逐个启动，观察现象
4.3 关闭其中执行的进程，观察现象
- Leader节点选举机制
每个Elastic-Job的任务执行实例App作为Zookeeper的客户端来操作ZooKeeper的znode
5.1 多个实例同时创建/leader节点
5.2 leader节点只能创建一个，后创建的会失败，创建成功的实例会被选为leader节点，执行任务
- Elastic-Job-Lite轻量级去中心化的特点
ElasticJob任务分片
ElasticJob可以把作业分为多个的task（每一个task就是一个任务分片）,每一个task交给具体的一个机器实例去处理（一个机器实例是可以处理多个task的），但是具体每个task执行什么逻辑由我们自己来指定.
