分布式调度
1. 什么是分布式调度
-
运⾏在分布式集群环境下的调度任务（同⼀个定时任务程序部署多份，只应该有⼀个定时任务在执⾏）
-
分布式调度—>定时任务的分布式—>定时任务的拆分（即为把⼀个⼤的作业任务拆分为多个⼩的作业任务，同时执⾏）
2. 定时任务与消息队列的区别
-
共同点
- 异步处理 ⽐如注册、下单事件
- 应⽤解耦 不管定时任务作业还是MQ都可以作为两个应⽤之间的⻮轮实现应⽤解耦，这个⻮轮可以中转数据，当然单体服务不需要考虑这些，服务拆分的时候往往都会考虑
- 流量削峰
-
不同点 定时任务作业是时间驱动，⽽MQ是事件驱动；
时间驱动是不可代替的，⽐如⾦融系统每⽇的利息结算，不是说利息来⼀条（利息到来事件）就算⼀下，⽽往往是通过定时任务批量计算；
所以，定时任务作业更倾向于批处理，MQ倾向于逐条处理；
3. 定时任务的实现⽅式
3.1 任务调度框架Quartz回顾
引入依赖
<dependency>
<groupId>org.quartz-scheduler</groupId>
<artifactId>quartz</artifactId>
<version>2.3.2</version>
</dependency>
自定义Job类
public class MyJob implements Job {
@Override
public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException {
System.out.println("执行myjob...");
}
}
定时任务作业主调度程序
public class QuartzMain {
/**
* 创建⼀个作业任务调度器
* @return
* @throws SchedulerException
*/
private static Scheduler createScheduler() throws SchedulerException {
SchedulerFactory schedulerFactory = new StdSchedulerFactory();
return schedulerFactory.getScheduler();
}
/**
* 创建⼀个作业任务
* MyJob.class 自定义Job类
* @return
*/
private static JobDetail createJob() {
JobBuilder jobBuilder = JobBuilder.newJob(MyJob.class);
jobBuilder.withIdentity("jobName", "myJob");
return jobBuilder.build();
}
/**
* 创建⼀个作业任务时间触发器
* 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
* 7、Year（年）1970~2099 可选项
* @return
*/
private static Trigger createTrigger() {
CronTrigger build = TriggerBuilder.newTrigger()
.withIdentity("triggerName", "myTrigger")
.startNow()
.withSchedule(CronScheduleBuilder.cronSchedule("0/2 * * * * ?"))
.build();
return build;
}
// 定时任务作业主调度程序
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();
}
}
4. 分布式调度框架Elastic-Job
4.1 Elastic-Job介绍
Elastic-Job是当当⽹开源的⼀个分布式调度解决⽅案，基于Quartz⼆次开发的，由两个相互独⽴的⼦项⽬Elastic-Job-Lite和Elastic-Job-Cloud组成。我们要学习的是 Elastic-Job-Lite，它定位为轻量级⽆中⼼化解决⽅案，使⽤Jar包的形式提供分布式任务的协调服务，⽽Elastic-Job-Cloud⼦项⽬需要结合Mesos以及Docker在云环境下使⽤。
Elastic-Job的github地址：https://github.com/elasticjob
主要功能介绍：
- 分布式调度协调 在分布式环境中，任务能够按指定的调度策略执⾏，并且能够避免同⼀任务多实例重复执⾏
- 丰富的调度策略 基于成熟的定时任务作业框架Quartz cron表达式执⾏定时任务
- 弹性扩容缩容 当集群中增加某⼀个实例，它应当也能够被选举并执⾏任务；当集群减少⼀个实例时，它所执⾏的任务能被转移到别的实例来执⾏。
- 失效转移 某实例在任务执⾏失败后，会被转移到其他实例执⾏
- 错过执⾏作业重触发 若因某种原因导致作业错过执⾏，⾃动记录错过执⾏的作业，并在上次作业完成后⾃动触发。
- ⽀持并⾏调度 ⽀持任务分⽚，任务分⽚是指将⼀个任务分为多个⼩任务项在多个实例同时执⾏。
- 作业分⽚⼀致性 当任务被分⽚后，保证同⼀分⽚在分布式环境中仅⼀个执⾏实例。
4.2 Elastic-Job-Lite应⽤
Elastic-Job依赖于Zookeeper进⾏分布式协调，所以需要安装Zookeeper软件（3.4.6版本以上）
安装zookepper
引入依赖
<!-- 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;
- 程序开发
- 定时任务类
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 action = (String) stringObjectMap.get("action"); String ip = (String) stringObjectMap.get("ip"); // 打印出这条记录 System.out.println("======>>>id：" + id + " action：" + action + " v：" + ip); // 更改状态 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); } }
- JdbcUtil
```java
public class JdbcUtil {
//url
private static String url = "jdbc:mysql://localhost:3306/ssm?characterEncoding=utf8&useSSL=false";
//user
private static String user = "root";
//password
private static String password = "root";
//驱动程序类
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;
}
}
- 程序主类
```java
public class ElasticJobMain {
public static void main(String[] args) {
// 配置注册中⼼zookeeper，zookeeper协调调度，不能让任务重复执⾏，通过命名空间分类管理任务，对应到zookeeper的⽬录
ZookeeperConfiguration zookeeperConfiguration = new ZookeeperConfiguration("127.0.0.1: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();
}
}
- 测试程序
1. 可先启动⼀个进程，然后再启动⼀个进程（两个进程模拟分布式环境下，通⼀个定时任务部署了两份在⼯作）
2. 两个进程逐个启动，观察现象
3. 关闭其中执⾏的进程，观察现象
##### 4.4 Elastic-Job-Lite轻量级去中⼼化的特点
##### 4.5 任务分⽚
⼀个⼤的⾮常耗时的作业Job，⽐如：⼀次要处理⼀亿的数据，那这⼀亿的数据存储在数据库中，如果⽤⼀个作业节点处理⼀亿数据要很久，在互联⽹领域是不太能接受的，互联⽹领域更希望机器的增加去横向扩展处理能⼒。所以，ElasticJob可以把作业分为多个的task（每⼀个task就是⼀个任务分⽚），每⼀个task交给具体的⼀个机器实例去处理（⼀个机器实例是可以处理多个task的），但是具体每个task执⾏什么逻辑由我们⾃⼰来指定。
Strategy策略定义这些分⽚项怎么去分配到各个机器上去，默认是平均去分，可以定制，⽐如某⼀个机器负载 ⽐较⾼或者预配置⽐较⾼，那么就可以写策略。分⽚和作业本身是通过⼀个注册中⼼协调的，因为在分布式环境下，状态数据肯定集中到⼀点，才可以在分布式中沟通。
```java
// 配置任务
// shardingTotalCount 分片数量
// shardingItemParameters 设置分片参数
JobCoreConfiguration jobCoreConfiguration = JobCoreConfiguration.newBuilder("archive-job", "*/2 * * * * ?", 3)
.shardingItemParameters("0=bachelor,1=master,2=doctor").build();
@Override
public void execute(ShardingContext shardingContext) {
int shardingItem = shardingContext.getShardingItem();
System.out.println("当前分片====>：" + shardingItem);
// 当前分片中的参数
String shardingParameter = shardingContext.getShardingParameter();
/**
* 从resume数据表查找1条未归档的数据，将其归档到resume_bak表，并更新状态为已归档（不删除原数据）
*/
// 查询出⼀条数据
String selectSql = "select * from resume where state='未归档' and education='" + shardingParameter + "' 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 action = (String) stringObjectMap.get("action");
String ip = (String)
stringObjectMap.get("ip");
// 打印出这条记录
System.out.println("======>>>id：" + id + " action：" +
action + " v：" + ip);
// 更改状态
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);
}
4.6 弹性扩容
新增加⼀个运⾏实例app3，它会⾃动注册到注册中⼼，注册中⼼发现新的服务上线，注册中⼼会通知ElasticJob 进⾏重新分⽚，那么总得分⽚项有多少，那么就可以搞多少个实例机器。
注意：
- 分⽚项也是⼀个JOB配置，修改配置，重新分⽚，在下⼀次定时运⾏之前会重新调⽤分⽚算法，那么这个分⽚算法的结果就是：哪台机器运⾏哪⼀个⼀⽚，这个结果存储到zk中的，主节点会把分⽚给分好放到注册中⼼去，然后执⾏节点从注册中⼼获取信息(执⾏节点在定时任务开启的时候获取相应的分⽚)。
- 如果所有的节点挂掉值剩下⼀个节点，所有分⽚都会指向剩下的⼀个节点，这也是ElasticJob的⾼可⽤。