分布式调度问题及解决方案

调度—>定时任务，分布式调度—>在分布式集群环境下定时任务这件事

Elastic-job(当当网开源的分布式调度框架)

定时任务的场景

定时任务形式:每隔一定时间/特定某一时刻执行例如:

订单审核、出库
订单超时自动取消、支付退款
礼券同步、生成、发放作业
物流信息推送、抓取作业、退换货处理作业
数据积压监控、日志监控、服务可用性探测作业
定时备份数据
金融系统每天的定时结算
数据归档、清理作业
报表、离线数据分析作业

什么是分布式调度

1)运行在分布式集群环境下的调度任务(同一个定时任务程序部署多份，只应该有一个定时任务在执行)

2)分布式调度—>定时任务的分布式—>定时任务的拆分(即为把一个大的作业任务拆分为多个小的作业任务，同时执行)

定时任务与消息队列的区别

共同点

异步处理

 比如注册、下单事件         应用解耦

不管定时任务作业还是MQ都可以作为两个应用之间的⻮轮实现应用解耦，这个⻮轮可以中转数据，当然单体服务不需要考虑这些，服务拆分的时候往往都会考虑

流量削峰

双十一的时候，任务作业和MQ都可以用来扛流量，后端系统根据服务能力定时处理订单或者从MQ抓取订单抓取到一个订单到来事件的话触发处理，对于前端用户来说看到的结果是已经下单成功了，下单是不受任何影响的

本质不同

定时任务作业是时间驱动，而MQ是事件驱动;

时间驱动是不可代替的，比如金融系统每日的利息结算，不是说利息来一条(利息到来事件)就算
一下，而往往是通过定时任务批量计算;

所以，定时任务作业更倾向于批处理，MQ倾向于逐条处理

定时任务的实现方式

定时任务的实现方式有多种。早期没有定时任务框架的时候，我们会使用JDK中的Timer机制和多线程机制(Runnable+线程休眠)来实现定时或者间隔一段时间执行某一段程序;后来有了定时任务框架，比如大名鼎鼎的Quartz任务调度框架，使用时间表达式(包括:秒、分、时、日、周、年)配置某一个任务什么时间去执行:

任务调度框架Quartz回顾

引入jar

<!-- 任务调度框架quartz -->
<dependency><groupId>org.quartz-scheduler</groupId><artifactId>quartz</artifactId><version>2.3.2</version>
</dependency>

定义job主程序

/*** @name: QuartzTest* @author: terwer* @date: 2022-02-08 17:55**/
public class QuartzTest {// 创建作业调度器(类似公交调度站)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* 7、Year(年)1970~2099 可选项* 示例:* 0 0 11 * * ? 每天的11点触发执行一次* 0 30 10 1 * ? 每月1号上午10点半触发执行一次**/public static Trigger createTrigger() {CronTrigger cronTrigger = 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 cronTrigger;}public static void main(String[] args) throws SchedulerException {// 创建一个作业任务调度器(类似于公交调度站)Scheduler scheduler = createScheduler();// 创建一个作业任务(类似于一辆公交⻋)JobDetail job = createJob();// 创建一个作业任务时间触发器(类似于公交⻋出⻋时间表)Trigger trigger = createTrigger();// 使用调度器按照时间触发器执行这个作业任务scheduler.scheduleJob(job, trigger);scheduler.start();}
}

定义一个job，需实现Job接口

/*** @name: DemoJob* @author: terwer* @date: 2022-02-08 18:23**/
public class DemoJob implements Job {public void execute(JobExecutionContext jobExecutionContext)throws JobExecutionException {System.out.println("这是一个测试的定时任务");}
}

效果

分布式调度框架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

备注：此项目目前已经捐赠以apache，新的项目地址是

https://github.com/apache/shardingsphere-elasticjob

最新版本是 3.0.1

主要功能介绍

分布式调度协调

在分布式环境中，任务能够按指定的调度策略执行，并且能够避免同一任务多实例重复执行

丰富的调度策略

基于成熟的定时任务作业框架Quartz cron表达式执行定时任务

弹性扩容缩容

当集群中增加某一个实例，它应当也能够被选举并执行任务;当集群减少一个实例时，它所执行的任务能被转移到别的实例来执行。

失效转移

某实例在任务执行失败后，会被转移到其他实例执行

错过执行作业重触发

若因某种原因导致作业错过执行，自动记录错过执行的作业，并在上次作业 完成后自动触发。

支持并行调度

支持任务分片，任务分片是指将一个任务分为多个小任务项在多个实例同时执行。 作业分片一致性 当任务被分片后，保证同一分片在分布式环境中仅一个执行实例。

Elastic-Job-Lite应用

jar包(API) + 安装zk软件

Elastic-Job依赖于Zookeeper进行分布式协调，所以需要安装Zookeeper软件(3.4.6版本以上)，Zookeeper的本质功能: 存储+通知。

安装Zookeeper(此处单例配置)

1)下载3.7.0版本，在linux平台解压下载的zookeeper-3.7.0-bin.tar.gz

https://dlcdn.apache.org/zookeeper/zookeeper-3.7.0/apache-zookeeper-3.7.0-bin.tar.gz源码
https://dlcdn.apache.org/zookeeper/zookeeper-3.7.0/apache-zookeeper-3.7.0.tar.gz

2)进入conf目录，cp zoo_sample.cfg zoo.cfg

3)进入bin目录，启动zk服务

启动 ./zkServer.sh start

停止 ./zkServer.sh stop

查看状态 ./zkServer.sh status

zookeeper树形节点结构图

引入elastic-job-lite jar包

<dependency><groupId>org.apache.shardingsphere.elasticjob</groupId><artifactId>elasticjob-lite-core</artifactId><version>3.0.1</version>
</dependency>

定时任务实例

需求:每隔两秒钟执行一次定时任务(resume表中未归档的数据归档到resume_bak表中，每次归档1条记录)

1)resume_bak和resume表结构完全一样

2)resume表中数据归档之后不删除，只将state置为"已归档"

数据表结构

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;

代码实现

1、定时任务类

/*** @name: BackupJob* @author: terwer* @date: 2022-02-08 22:06**/
public class BackupJob implements SimpleJob {// 定时任务每执行一次都会执行如下的逻辑@Overridepublic 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);}
}

2、主类

/*** @name: ElasticJobMain* @author: terwer* @date: 2022-02-08 22:09**/
public class ElasticJobMain {public static void main(String[] args) {// 配置注册中心zookeeper，zookeeper协调调度，不能让任务重复执行， 通过命名空间分类管理任务，对应到zookeeper的目录ZookeeperConfiguration zookeeperConfiguration = newZookeeperConfiguration("localhost:2181", "data-archive-job");CoordinatorRegistryCenter coordinatorRegistryCenter = newZookeeperRegistryCenter(zookeeperConfiguration);coordinatorRegistryCenter.init();// 配置任务// 参考 https://shardingsphere.apache.org/elasticjob/current/en/quick-start/elasticjob-lite/JobConfiguration jobConfiguration = JobConfiguration.newBuilder("archive-job", 1).cron("*/2 * * * * ? ").build();// 新版本下面写法失效// JobCoreConfiguration JobConfiguration =//        JobCoreConfiguration.newBuilder("archive-job", "*/2 * * * * ? ",1).build();//                SimpleJobConfiguration simpleJobConfiguration = new//                        SimpleJobConfiguration(jobCoreConfiguration, BackupJob.class.getName());//// 启动任务new ScheduleJobBootstrap(coordinatorRegistryCenter, new BackupJob(), jobConfiguration).schedule();// 新版本下面写法失效// new JobScheduler(coordinatorRegistryCenter,//        LiteJobConfiguration.newBuilder(simpleJobConfiguration).build()).init();}
}

3、JdbcUtil工具类

/*** @name: JdbcUtil* @author: terwer* @date: 2022-02-08 22:11**/
public class JdbcUtil {//urlprivate static String url = "jdbc:mysql://localhost:3306/job?characterEncoding=utf8&useSSL=false";//userprivate static String user = "root"; //passwordprivate static String password = "123456"; //驱动程序类private static String driver = "com.mysql.jdbc.Driver";static {try {Class.forName(driver);} catch (ClassNotFoundException e) {e.printStackTrace();}}public static Connection getConnection() {try {return DriverManager.getConnection(url, user, password);} catch (SQLException e) {e.printStackTrace();}return null;}public static void close(ResultSet rs, PreparedStatement ps,Connection con) {if (rs != null) {try {rs.close();} catch (SQLException e) {e.printStackTrace();} finally {if (ps != null) {try {ps.close();} catch (SQLException e) {e.printStackTrace();} finally {if (con != null) {try {con.close();} catch (SQLException e) {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) {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) {e.printStackTrace();} finally {close(rs, ps, con);}return null;}
}

测试

1)可先启动一个进程，然后再启动一个进程(两个进程模拟分布式环境下，通一个定时任务部署了两份在工作)

2)两个进程逐个启动，观察现象

3)关闭其中执行的进程，观察现象

Leader节点选举机制

每个Elastic-Job的任务执行实例App作为Zookeeper的客户端来操作ZooKeeper的znode

(1)多个实例同时创建/leader节点

(2)/leader节点只能创建一个，后创建的会失败，创建成功的实例会被选为leader节点，执行任务

备注：
zk启动失败的几个可能原因

1、端口被占用

lsof -i tcp:8080

2、zoo.cfg的数据文件和日志文件目录问题

远程链接如果失败，可以关闭防火墙

sudo service firewalld stop

Elastic-Job-Lite轻量级去中心化的特点

任务分片

一个大的非常耗时的作业Job，比如:一次要处理一亿的数据，那这一亿的数据存储在数据库中，如果用一个作业节点处理一亿数据要很久，在互联网领域是不太能接受的，互联网领域更希望机器的增加去横向扩展处理能力。所以，ElasticJob可以把作业分为多个的task(每一个task就是一个任务分片)，每一个task交给具体的一个机器实例去处理(一个机器实例是可以处理多个task的)，但是具体每个task 执行什么逻辑由我们自己来指定。

Strategy策略定义这些分片项怎么去分配到各个机器上去，默认是平均去分，可以定制，比如某一个机器负载比较高或者预配置比较高，那么就可以写策略。分片和作业本身是通过一个注册中心协调的，因为在分布式环境下，状态数据肯定集中到一点，才可以在分布式中沟通。

弹性扩容

新增加一个运行实例app3，它会自动注册到注册中心，注册中心发现新的服务上线，注册中心会通知 ElasticJob 进行重新分片，那么总得分片项有多少，那么就可以搞多少个实例机器，比如完全可以分 1000片

最多就可以有多少app实例机器能成的主，完全可以分1000片那么就可以搞1000台机器一起执行作业

注意:
1)分片项也是一个JOB配置，修改配置，重新分片，在下一次定时运行之前会重新调用分片算法，那么这个分片算法的结果就是:哪台机器运行哪一个一片，这个结果存储到zk中的，主节点会把分片给分好放到注册中心去，然后执行节点从注册中心获取信息(执行节点在定时任务开启的时候获取相应的分片)。

2)如果所有的节点挂掉值剩下一个节点，所有分片都会指向剩下的一个节点，这也是ElasticJob的高可用。