一、對分布式調度的理解
Elastic-job（當當⽹開源的分布式調度框架）
定時任務形式：每隔⼀定時間/特定某⼀時刻執⾏ 例如：
訂單審核、出庫 訂單超時⾃動取消、⽀付退款 禮券同步、⽣成、發放作業 物流信息推送、抓取作業、退換貨處理作業
數據積壓監控、⽇志監控、服務可⽤性探測作業 定時備份數據
⾦融系統每天的定時結算 數據歸檔、清理作業 報表、離線數據分析作業
2 什么是分布式調度
什么是分布式任務調度？有兩層含義
1）運⾏在分布式集群環境下的調度任務（同⼀個定時任務程序部署多份，只應該有⼀個定時任務在執
⾏）
3、分布式調度Elastic-Job與zookeeperk
特點優點
-
輕量級去中⼼化
1、ElasticJob可以把作業分為多個的task（每⼀個task就是⼀個任務分⽚），每⼀個task交給具體的⼀個機器2、實例去處理（⼀個機器實例是可以處理多個task的），但是具體每個task 執⾏什么邏輯由我們⾃⼰來指定。
3、默認是平均去分，可以定制。分⽚項也是⼀個JOB配置，修改配置，重新分⽚，在下⼀次定時運⾏之前會重新調⽤分⽚算法
結果就是：哪台機器運⾏哪⼀個⼀⽚，這個結果存儲到zookeeperk中的，主節點會把分⽚給分好 放到注冊中⼼去，然后執⾏節點從注冊中⼼獲取信息(執⾏節點在定時任務開啟的時候獲取相應的分⽚
2）如果所有的節點掛掉值剩下⼀個節點，所有分⽚都會指向剩下的⼀個節點，這也是ElasticJob的⾼可
3、 彈性擴容
總結：
分布式調度ElasticJob目的是解決某一個job節點的服務器壓力（一個人做，和一堆人分工去做的）利用zookeeperk 輕量級去中⼼、任務分⽚、彈性擴容 三大特點，實現分片分工。快速有效、協調完成工作。不會出現分片重復工作的情況。
二、准備驗證環境
1、安裝zookeeper
https://www.cnblogs.com/aGboke/p/12904932.html
zooInspector的使用： https://www.cnblogs.com/lwcode6/p/11586537.html
elastic-job：https://github.com/elasticjob
2、搭建maven項目、引入
<!--數據庫驅動jar-->
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>5.1.46</version>
</dependency>
<!--任務調度框架quartz-->
<!--org.quartz-scheduler/quartz -->
<dependency>
<groupId>org.quartz-scheduler</groupId>
<artifactId>quartz</artifactId>
<version>2.3.2</version>
</dependency>
<!--elastic-job-lite-core-->
<dependency>
<groupId>com.dangdang</groupId>
<artifactId>elastic-job-lite-core</artifactId>
<version>2.1.5</version>
</dependency>
3、測試代碼
package com.lagou.job; import com.dangdang.ddframe.job.api.ShardingContext; import com.dangdang.ddframe.job.api.simple.SimpleJob; 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; import java.util.List; import java.util.Map; /** * @author Mrwg * @date 2020/5/15 * @description */ 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); } public static void main(String[] args) { //配置分布式Zookeeper分布式協調中心 ZookeeperConfiguration zookeeperConfiguration = new ZookeeperConfiguration("ip:2181", "elastic-job"); CoordinatorRegistryCenter coordinatorRegistryCenter = new ZookeeperRegistryCenter(zookeeperConfiguration); coordinatorRegistryCenter.init(); //配置任務 每秒運行一次 JobCoreConfiguration jobCoreConfiguration = JobCoreConfiguration.newBuilder("archive-job", "1 * * * * ?", 1).build(); SimpleJobConfiguration simpleJobConfiguration = new SimpleJobConfiguration(jobCoreConfiguration, BackupJob.class.getName()); //啟動任務 new JobScheduler(coordinatorRegistryCenter, LiteJobConfiguration.newBuilder(simpleJobConfiguration).build()).init(); } }
package com.lagou.job; import java.sql.*; import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; /** * @author Mrwg * @date 2020/5/15 * @description */ public class JdbcUtil { //url private static String url = "jdbc:mysql://localhost:3306/test?characterEncoding=utf8&useSSL=false"; //user private static String user = ""; //password private static String password = ""; //驅動程序類 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; } }
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`) USING BTREE ) ENGINE=InnoDB AUTO_INCREMENT=0 DEFAULT CHARSET=utf8; create table resume_bak like resume; INSERT INTO `test`.`resume`(`id`, `name`, `sex`, `phone`, `address`, `education`, `state`) VALUES (1, '2', 'girl', '18801240649', '北京', '本科', '已歸檔'); INSERT INTO `test`.`resume`(`id`, `name`, `sex`, `phone`, `address`, `education`, `state`) VALUES (2, '2', 'girl2', '18801240649', '北京', '本科', '已歸檔'); INSERT INTO `test`.`resume`(`id`, `name`, `sex`, `phone`, `address`, `education`, `state`) VALUES (3, '3', 'girl3', '18801240649', '北京', '本科', '已歸檔'); INSERT INTO `test`.`resume`(`id`, `name`, `sex`, `phone`, `address`, `education`, `state`) VALUES (4, '4', 'girl4', '18801240649', '北京', '本科', '已歸檔'); INSERT INTO `test`.`resume`(`id`, `name`, `sex`, `phone`, `address`, `education`, `state`) VALUES (5, '5', 'girl5', '18801240649', '北京', '本科', '已歸檔'); INSERT INTO `test`.`resume`(`id`, `name`, `sex`, `phone`, `address`, `education`, `state`) VALUES (6, '6', 'girl6', '18801240649', '北京', '本科', '已歸檔'); INSERT INTO `test`.`resume`(`id`, `name`, `sex`, `phone`, `address`, `education`, `state`) VALUES (7, '7', 'girl7', '18801240649', '北京', '本科', '已歸檔'); INSERT INTO `test`.`resume`(`id`, `name`, `sex`, `phone`, `address`, `education`, `state`) VALUES (8, '8', 'girl8', '18801240649', '北京', '本科', '已歸檔'); INSERT INTO `test`.`resume`(`id`, `name`, `sex`, `phone`, `address`, `education`, `state`) VALUES (9, '9', 'girl9', '18801240649', '北京', '本科', '已歸檔'); INSERT INTO `test`.`resume`(`id`, `name`, `sex`, `phone`, `address`, `education`, `state`) VALUES (10, '10', 'girl10', '18801240649', '北京', '本科', '已歸檔'); INSERT INTO `test`.`resume`(`id`, `name`, `sex`, `phone`, `address`, `education`, `state`) VALUES (11, '11', 'girl11', '18801240649', '北京', '本科', '已歸檔'); INSERT INTO `test`.`resume`(`id`, `name`, `sex`, `phone`, `address`, `education`, `state`) VALUES (12, '12', 'girl12', '18801240649', '北京', '本科', '已歸檔'); INSERT INTO `test`.`resume`(`id`, `name`, `sex`, `phone`, `address`, `education`, `state`) VALUES (13, '13', 'girl13', '18801240649', '北京', '本科', '已歸檔'); INSERT INTO `test`.`resume`(`id`, `name`, `sex`, `phone`, `address`, `education`, `state`) VALUES (14, '14', 'girl14', '18801240649', '北京', '本科', '已歸檔'); INSERT INTO `test`.`resume`(`id`, `name`, `sex`, `phone`, `address`, `education`, `state`) VALUES (15, '15', 'girl15', '18801240649', '北京', '本科', '已歸檔'); INSERT INTO `test`.`resume`(`id`, `name`, `sex`, `phone`, `address`, `education`, `state`) VALUES (16, '16', 'girl16', '18801240649', '北京', '本科', '已歸檔'); INSERT INTO `test`.`resume`(`id`, `name`, `sex`, `phone`, `address`, `education`, `state`) VALUES (17, '17', 'girl17', '18801240649', '北京', '本科', '已歸檔'); INSERT INTO `test`.`resume`(`id`, `name`, `sex`, `phone`, `address`, `education`, `state`) VALUES (18, '18', 'girl18', '18801240649', '北京', '本科', '已歸檔'); INSERT INTO `test`.`resume`(`id`, `name`, `sex`, `phone`, `address`, `education`, `state`) VALUES (19, '19', 'girl19', '18801240649', '北京', '本科', '已歸檔'); INSERT INTO `test`.`resume`(`id`, `name`, `sex`, `phone`, `address`, `education`, `state`) VALUES (20, '20', 'girl20', '18801240649', '北京', '本科', '已歸檔'); INSERT INTO `test`.`resume`(`id`, `name`, `sex`, `phone`, `address`, `education`, `state`) VALUES (21, '21', 'girl21', '18801240649', '北京', '本科', '已歸檔'); INSERT INTO `test`.`resume`(`id`, `name`, `sex`, `phone`, `address`, `education`, `state`) VALUES (22, '22', 'girl22', '18801240649', '北京', '本科', '已歸檔'); INSERT INTO `test`.`resume`(`id`, `name`, `sex`, `phone`, `address`, `education`, `state`) VALUES (23, '23', 'girl23', '18801240649', '北京', '本科', '已歸檔'); INSERT INTO `test`.`resume`(`id`, `name`, `sex`, `phone`, `address`, `education`, `state`) VALUES (24, '24', 'girl24', '18801240649', '北京', '本科', '已歸檔');
4、啟動main()方法 ，zooInspector 鏈接 zookeeper
1、啟動一個實列
當前定時任務，全部在當前實列下執行，啟動倆個實列，zk會重新計算分片和競爭機制，來確定那台機器運行當前任務。（一般情況下第二個實列會拿到領導權），當我們把倆個實列，
其中一個停掉，第一個實列會繼續接着運行未完成的任務。 如下下邊gif所示。運行速度受當前網絡、機器硬件影響。
2、啟動倆個實列
3、調整分片數量
- 3個分片，啟動1個main()方法（如下所示gif）
JobCoreConfiguration jobCoreConfiguration = JobCoreConfiguration.newBuilder("archive-job", "1 * * * * ?", 3).build();// 任務名稱 執行時間 分片數
- 3個分片2個main()
- 3個分片3個main()實列
