分布式集群架构场景化解决⽅案（四）分布式调度问题_dangdang simplejob

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

Elastic-job （当当⽹开源的分布式调度框架）

定时任务形式：每隔⼀定时间 / 特定某⼀时刻执⾏

例如：

二. 什么是分布式调度

什么是分布式任务调度？有两层含义

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

（一）共同点

（二）本质不同

四. 定时任务的实现⽅式

（一）任务调度框架Quartz回顾示意(我们课程主要内容不是Quartz)

1. 引入jar

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

2. 定时任务作业主调度程序

package quartz;
 
import org.quartz.*;
import org.quartz.impl.StdSchedulerFactory;
 
public class QuartzMan {
 
    // 1、创建任务调度器（好比公交调度站）
    public static Scheduler createScheduler() throws SchedulerException {
        SchedulerFactory schedulerFactory = new StdSchedulerFactory();
        Scheduler scheduler = schedulerFactory.getScheduler();
        return scheduler;
    }
 
 
    // 2、创建一个任务（好比某一个公交车的出行）
    public static JobDetail createJob() {
        JobBuilder jobBuilder = JobBuilder.newJob(DemoJob.class); // TODO 自定义任务类
        //设置任务的基本属性
        jobBuilder.withIdentity("jobName","myJob");
        JobDetail jobDetail = jobBuilder.build();
        return jobDetail;
    }
 
 
    /**
     * 3、创建作业任务时间触发器（类似于公交车出车时间表）
     * 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()
                //设置时间表达式：每隔2秒执行一次
                .withSchedule(CronScheduleBuilder.cronSchedule("*/2 * * * * ?")).build();
        return cronTrigger;
    }
 
 
 
    /**
     * main函数中开启定时任务
     * @param args
     */
    public static void main(String[] args) throws SchedulerException {
        // 1、创建任务调度器（好比公交调度站）
        Scheduler scheduler = QuartzMan.createScheduler();
        // 2、创建一个任务（好比某一个公交车的出行）
        JobDetail job = QuartzMan.createJob();
        // 3、创建任务的时间触发器（好比这个公交车的出行时间表）
        Trigger trigger = QuartzMan.createTrigger();
        // 4、使用任务调度器根据时间触发器执行我们的任务
        scheduler.scheduleJob(job,trigger);
        scheduler.start();
    }
}

3. 定义⼀个job，需实现Job接⼝

package quartz;
 
import org.quartz.Job;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
 
public class DemoJob implements Job {
    @Override
    public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException {
        System.out.println("我是一个定时任务执行逻辑");
    }
}

五. 分布式调度框架Elastic-Job

（一）Elastic-Job介绍

• 分布式调度协调
• 在分布式环境中，任务能够按指定的调度策略执⾏，并且能够避免同⼀任务多实例重复执⾏
• 丰富的调度策略 基于成熟的定时任务作业框架Quartz cron表达式执⾏定时任务
• 弹性扩容缩容 当集群中增加某⼀个实例，它应当也能够被选举并执⾏任务；当集群减少⼀个实例时，它所执⾏的任务能被转移到别的实例来执⾏。
• 失效转移 某实例在任务执⾏失败后，会被转移到其他实例执⾏
• 错过执⾏作业重触发 若因某种原因导致作业错过执⾏，⾃动记录错过执⾏的作业，并在上次作业完成后⾃动触发。
• ⽀持并⾏调度 ⽀持任务分⽚，任务分⽚是指将⼀个任务分为多个⼩任务项在多个实例同时执⾏。
• 作业分⽚⼀致性 当任务被分⽚后，保证同⼀分⽚在分布式环境中仅⼀个执⾏实例

（二）Elastic-Job-Lite应⽤

1. 安装Zookeeper（此处单例配置）

 3. 引入jar包

<!-- https://mvnrepository.com/artifact/com.dangdang/elastic-job-lite-core -->
        <!--elastic-job-lite核心包-->
        <dependency>
            <groupId>com.dangdang</groupId>
            <artifactId>elastic-job-lite-core</artifactId>
            <version>2.1.5</version>
        </dependency>

4. 定时任务实例

-- ----------------------------
-- 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;

5. 定时任务类

package elasticjob;
 
import com.dangdang.ddframe.job.api.ShardingContext;
import com.dangdang.ddframe.job.api.simple.SimpleJob;
 
import java.util.List;
import java.util.Map;
 
/**
 * ElasticJobLite定时任务业务逻辑处理类
 */
public class ArchivieJob implements SimpleJob {
 
    /**
     * 需求：resume表中未归档的数据归档到resume_bak表中，每次归档1条记录
     * execute方法中写我们的业务逻辑（execute方法每次定时任务执行都会执行一次）
     * @param shardingContext
     */
    @Override
    public void execute(ShardingContext shardingContext) {
        //获取当前的分片
        int shardingItem = shardingContext.getShardingItem();
        System.out.println("=====>>>>当前分片：" + shardingItem);
 
        // 获取分片参数shardingItemParameters("0=bachelor,1=master,2=doctor").build();在ElasticJobMain类中设置的
        String shardingParameter = shardingContext.getShardingParameter(); // 0=bachelor,1=master,2=doctor
 
 
        // 1 从resume表中查询出1条记录（未归档）
        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 ) {
            System.out.println("数据已经处理完毕！！！！！！");
            return;
        }
        // 2 "未归档"更改为"已归档"
        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);
 
        // 3 归档这条记录，把这条记录插入到resume_bak表
        String insertSql = "insert into resume_bak select * from resume where id=?";
        JdbcUtil.executeUpdate(insertSql,id);
    }
}

6. 主类

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
        ZookeeperConfiguration zookeeperConfiguration = new ZookeeperConfiguration("192.168.200.128:2181","data-archive-job");
        CoordinatorRegistryCenter coordinatorRegistryCenter = new ZookeeperRegistryCenter(zookeeperConfiguration);
        coordinatorRegistryCenter.init();
 
        // 配置任务（时间事件、定时任务业务逻辑、调度器）
        JobCoreConfiguration jobCoreConfiguration = JobCoreConfiguration
                .newBuilder("archive-job", "*/2 * * * * ?", 3)
                .shardingItemParameters("0=bachelor,1=master,2=doctor").build();
        SimpleJobConfiguration simpleJobConfiguration = new SimpleJobConfiguration(jobCoreConfiguration,ArchivieJob.class.getName());
 
        //这里加overwrite(true)的时候当重写上面的配置信息时会重写进zookeeper中
        JobScheduler jobScheduler = new JobScheduler(coordinatorRegistryCenter, LiteJobConfiguration.newBuilder(simpleJobConfiguration).overwrite(true).build());
        jobScheduler.init();
 
 
    }
}

7. JdbcUtil⼯具类

package elasticjob;
 
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://192.168.200.128:3306/jpa?characterEncoding=utf8&useSSL=false";
    //user
    private static String user = "root";
    //password
    private static String password = "PcPc@123";
    //驱动程序类
    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();
                            }
                        }
                    }
                }
            }
        }
    }
 
 
    /***
     * DML操作（增删改）
     * 1.获取连接数据库对象
     * 2.预处理
     * 3.执行更新操作
     * @param sql
     * @param obj
     */
    //调用者只需传入一个sql语句，和一个Object数组。该数组存储的是SQL语句中的占位符
    public static void executeUpdate(String sql,Object...obj) {
        Connection con = getConnection();//调用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);//调用close()方法关闭资源
        }
    }
 
 
 
    /***
     * DQL查询
     * Result获取数据集
     *
     * @param sql
     * @param obj
     * @return
     */
    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();
            //new 一个空的list集合用来存放查询结果
            List<Map<String, Object>> list = new ArrayList<>();
            //获取结果集的列数
            int count = rs.getMetaData().getColumnCount();
            //对结果集遍历每一条数据是一个Map集合，列是k,值是v
            while (rs.next()) {
                //一个空的map集合，用来存放每一行数据
                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);//获取k即列名
                    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;
    }
}

（四）Elastic-Job-Lite轻量级去中⼼化的特点

如何理解轻量级和去中⼼化？

 （五）任务分⽚

 

 

 

（六）弹性扩容