某电商平台需要每天上午10点下午3点晚上8点发放一批优惠券某银行系统需要在信用卡到期还款日的前三天进行短信提醒某财务系统需要在每天凌晨0:10分结算前一天的财务数据统计汇总

任务调度是为了自动完成特定任务在约定的特定时刻去执行任务的过程

Spring

感觉Spring给我们提供的这个注解可以完成任务调度的功能好像已经完美解决问题了为什么还需要分布式呢?

主要有如下这几点原因:

1.单机处理极限原本1分钟内需要处理1万个订单但是现在需要1分钟内处理10万个订单原来一个统计需要1小时现在业务方需要10分钟就统计出来。

你也许会说你也可以多线程、单机多进程处理。

的确多线程并行处理可以提高单位时间的处理效率但是单机能力毕竟有限主要是CPU、内存和磁盘始终会有单机处理不过来的情况。

2.高可用单机版的定式任务调度只能在一台机器上运行如果程序或者系统出现异常就会导致功能不可用。

虽然可以在单机程序实现的足够稳定但始终有机会遇到非程序引起的故障而这个对于一个系统的核心功能来说是不可接受的。

3.防止重复执行:

在单机模式下定时任务是没什么问题的。

但当我们部署了多台服务同时又每台服务又有定时任务时若不进行合理的控制在同一时间只有一个定时任务启动执行这时定时执行的结果就可能存在混乱和错误了

1.3

是一个分布式调度的解决方案由当当网开源它由两个相互独立的子项目

Elastic-job-Lite

https://shardingsphere.apache.org/elasticjob/

功能列表:

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

弹性拓容缩容

当集群中增加一个实例它应当能够被选举被执行任务当集群减少一个实例时他所执行的任务能被转移到别的示例中执行。

失效转移

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

支持并行调度

支持任务分片任务分片是指将一个任务分成多个小任务在多个实例同时执行。

作业分片一致性

dependencygroupIdcom.dangdang/groupIdartifactIdelastic-job-lite-core/artifactIdversion2.1.5/version

/dependency2.2

JobScheduler(createRegistryCenter(),

createJobConfiguration()).init();}private

static

{//配置zk地址,调度任务的组名ZookeeperConfiguration

new

ZookeeperConfiguration(localhost:2181,

elastic-job-demo);zookeeperConfiguration.setSessionTimeoutMilliseconds(100);CoordinatorRegistryCenter

regCenter

ZookeeperRegistryCenter(zookeeperConfiguration);regCenter.init();return

regCenter;}private

JobCoreConfiguration.newBuilder(demoSimpleJob,0/3

?,1).build();//

定义SIMPLE类型配置SimpleJobConfiguration

simpleJobConfig

SimpleJobConfiguration(simpleCoreConfig,

MyElasticJob.class.getCanonicalName());//

定义Lite作业根配置LiteJobConfiguration

simpleJobRootConfig

LiteJobConfiguration.newBuilder(simpleJobConfig).build();return

}2.2.3

运行单个程序查看是否按照cron表达式的内容进行任务的调度运行多个程序查看是否只会有一个实例进行任务调度运行多个程序后把正在进行任务调度的进程关掉查看其它进程是否能继续进行任务调度

3.1

groupIdcn.wolfcode/groupIdartifactIdelastic-job-springboot/artifactIdversion1.0.0/versionparentgroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-parent/artifactIdversion2.1.3.RELEASE/version/parentpropertiesproject.build.sourceEncodingUTF-8/project.build.sourceEncodingproject.reporting.outputEncodingUTF-8/project.reporting.outputEncodingjava.version1.8/java.version/propertiesdependenciesdependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-web/artifactId/dependencydependencygroupIdcom.dangdang/groupIdartifactIdelastic-job-lite-spring/artifactIdversion2.1.5/version/dependencydependencygroupIdorg.projectlombok/groupIdartifactIdlombok/artifactId/dependency/dependencies3.2

相关配置

因为配置中心的地址并不是固定的所以我们应该把这个地址信息配置在配置文件中所以在配置文件application.yml中添加配置如下:

Configuration

createRegistryCenter(Value(${elasticjob.zookeeper-url})

String

zookeeperUrl,Value(${elasticjob.group-name})

String

ZookeeperConfiguration(zookeeperUrl,groupName);//设置zk超时时间zookeeperConfiguration.setSessionTimeoutMilliseconds(100);//创建注册中心CoordinatorRegistryCenter

new

ZookeeperRegistryCenter(zookeeperConfiguration);return

}任务调度配置类

定义作业核心配置JobCoreConfiguration.Builder

JobCoreConfiguration.newBuilder(jobClass.getSimpleName(),

cron,

shardingTotalCount);if(!StringUtils.isEmpty(shardingItemParameters)){jobCoreConfigurationBuilder.shardingItemParameters(shardingItemParameters);}//

定义SIMPLE类型配置SimpleJobConfiguration

simpleJobConfig

SimpleJobConfiguration(jobCoreConfigurationBuilder.build(),

MyElasticJob.class.getCanonicalName());//

定义Lite作业根配置LiteJobConfiguration

simpleJobRootConfig

LiteJobConfiguration.newBuilder(simpleJobConfig).overwrite(true).build();return

simpleJobRootConfig;}Bean(initMethod

init)public

initSimpleElasticJob(){SpringJobScheduler

springJobScheduler

SpringJobScheduler(myElasticJob,registryCenter,createJobConfiguration(jobClass,0/3

?,1,null));return

需求数据库中有一些列的数据需要对这些数据进行备份操作备份完之后修改数据的状态标记已经备份了。

4.1

dependencygroupIdcom.alibaba/groupIdartifactIddruid/artifactIdversion1.1.10/version/dependencydependencygroupIdorg.mybatis.spring.boot/groupIdartifactIdmybatis-spring-boot-starter/artifactIdversion1.2.0/version/dependency!--mysql驱动--dependencygroupIdmysql/groupIdartifactIdmysql-connector-java/artifactId/dependency4.2.2

添加配置

jdbc:mysql://localhost:3306/elastic-job-demo?serverTimezoneGMT%2B8driverClassName:

com.alibaba.druid.pool.DruidDataSourceusername:

rootpassword:

fileCustomMapper;Overridepublic

void

fileCustomMapper.selectAll();System.out.println(需要备份文件个数:fileList.size());for(FileCustom

fileCustom:fileList){backUpFile(fileCustom);}}private

void

{//模拟备份动作TimeUnit.SECONDS.sleep(1);}

catch

{e.printStackTrace();}System.out.println(执行文件备份fileCustom);fileCustomMapper.changeState(fileCustom.getId(),1);}

}4.3.2

initFileCustomElasticJob(FileCustomElasticJob

fileCustomElasticJob){SpringJobScheduler

springJobScheduler

SpringJobScheduler(fileCustomElasticJob,registryCenter,createJobConfiguration(FileCustomElasticJob.class,0

0/1

为了高可用我们会对这个项目做集群的操作,可以保证其中一台挂了,另外一台可以继续工作.但是在集群的情况下,调度任务只在一台机器上运行如果单个任务调度比较耗时耗资源的情况下对这台机器的消耗还是比较大的

但是这个时候其他机器却是空闲着的.如何合理的利用集群的其他机器且如何让任务执行得更快些呢这时候Elastic-Job提供了任务调度分片的功能.

5.分片概念

作业分片是指任务的分布式执行需要将一个任务拆分为多个独立的任务项然后由分布式的应用实例分别执行某一个或者几个分布项。

例如Elastic-Job快速入门中文件备份的案例现有两台服务器每台服务器分别跑一个应用实例。

为了快速执行作业那么可以讲任务分成4片每个应用实例都执行两片。

作业遍历数据逻辑应为实例1查找text和image类型文件执行备份实例2查找radio和vedio类型文件执行备份。

如果由于服务器拓容应用实例数量增加为4则作业遍历数据的逻辑应为:

4个实例分别处理text,image,radio,video类型的文件。

分片项与业务处理解耦

Elastic-Job并不直接提供数据处理的功能,框架只会将分片项分配至各个运行中的作业服务器开发者需要自行处理分片项与真实数据的对应关系

最大限度利用资源

将分片项设置大于服务器的数据最好是大于服务器倍数的数量作业将会合理利用分布式资源动态的分配分片项.

例如:

3台服务器分成10片则分片项结果为服务器A0,1,2;服务器B3,4,5;服务器C6,7,8,9.如果

服务器C奔溃则分片项分配结果为服务器A0,1,2,3,4;服务器B5,6,7,8,9.在不丢失分片项的情况下最大限度利用现有的资源提高吞吐量.

6.案例改造成任务分片

initFileCustomElasticJob(FileCustomElasticJob

fileCustomElasticJob){SpringJobScheduler

springJobScheduler

SpringJobScheduler(fileCustomElasticJob,registryCenter,createJobConfiguration(FileCustomElasticJob.class,0

0/1

?,4,0text,1image,2radio,3vedio));return

新增作业分片逻辑

fileCustomMapper;Overridepublic

void

{doWork(shardingContext.getShardingParameter());}private

void

fileCustomMapper.selecByType(fileType);System.out.println(类型为:fileType,文件需要备份个数:fileList.size());for(FileCustom

fileCustom:fileList){backUpFile(fileCustom);}}private

void

{//模拟备份动作TimeUnit.SECONDS.sleep(1);}

catch

{e.printStackTrace();}System.out.println(执行文件备份fileCustom);fileCustomMapper.changeState(fileCustom.getId(),1);}

}6.3

只有一台机器的情况下任务分片是如何执行的有多台机器的情况下任务分片是如何执行的

7.Dataflow

接口该接口提供2个方法供覆盖分别用于抓取fetchData和处理processData数据我们继续对例子进行改造。

Dataflow类型用于处理数据流他和SimpleJob不同它以数据流的方式执行调用fetchData抓取数据知道抓取不到数据才停止作业。

7.1

fileCustomMapper;Overridepublic

ListFileCustom

fileCustomMapper.fetchData(2);System.out.println(抓取时间:new

Date(),个数fileCustoms.size());return

void

fileCustom:data){backUpFile(fileCustom);}}private

void

{//模拟备份动作TimeUnit.SECONDS.sleep(1);}

catch

{e.printStackTrace();}System.out.println(执行文件备份fileCustom);fileCustomMapper.changeState(fileCustom.getId(),1);}

}7.2

定义作业核心配置JobCoreConfiguration.Builder

JobCoreConfiguration.newBuilder(jobClass.getSimpleName(),

cron,

shardingTotalCount);if(!StringUtils.isEmpty(shardingItemParameters)){jobCoreConfigurationBuilder.shardingItemParameters(shardingItemParameters);}JobTypeConfiguration

jobConfig

null;if(dataflowType){jobConfig

new

DataflowJobConfiguration(jobCoreConfigurationBuilder.build(),jobClass.getCanonicalName(),true);}else{//

new

SimpleJobConfiguration(jobCoreConfigurationBuilder.build(),

jobClass.getCanonicalName());}//

定义Lite作业根配置LiteJobConfiguration

simpleJobRootConfig

LiteJobConfiguration.newBuilder(jobConfig).overwrite(true).build();return

simpleJobRootConfig;}/*Bean(initMethod

init)public

initSimpleElasticJob(){SpringJobScheduler

springJobScheduler

SpringJobScheduler(myElasticJob,registryCenter,createJobConfiguration(MyElasticJob.class,0/3

?,1,null));return

springJobScheduler;}*//*Bean(initMethod

init)public

initFileCustomElasticJob(FileCustomElasticJob

fileCustomElasticJob){SpringJobScheduler

springJobScheduler

SpringJobScheduler(fileCustomElasticJob,registryCenter,createJobConfiguration(FileCustomElasticJob.class,0

0/1

?,4,0text,1image,2radio,3vedio,false));return

springJobScheduler;}*/Bean(initMethod

init)public

iniDataflowElasticJob(FileDataflowJob

fileDataflowJob){SpringJobScheduler

springJobScheduler

SpringJobScheduler(fileDataflowJob,registryCenter,createJobConfiguration(FileDataflowJob.class,0

0/1

Elastic-Job-Lite在配置中提供了JobEventConfiguration,支持数据库方式配置会在数据库中自动创建JOB_EXECUTION_LOG和JOB_STATUS_TRACE_LOG两张表以及若干索引来近路作业的相关信息。

8.1.1

在ElasticJobConfig配置类中注入DataSource

Configuration

initFileCustomElasticJob(FileCustomElasticJob

fileCustomElasticJob){//增加任务事件追踪配置JobEventConfiguration

new

JobEventRdbConfiguration(dataSource);SpringJobScheduler

springJobScheduler

SpringJobScheduler(fileCustomElasticJob,registryCenter,createJobConfiguration(FileCustomElasticJob.class,0

0/1

?,4,0text,1image,2radio,3vedio,false),jobEventConfiguration);return

日志信息表

启动后会发现在elastic-job-demo数据库中新增以下两张表

job_execution_log

2.作业完成执行时向数据库更新数据更新is_success,complete_time和failure_cause(如果任务执行失败)

job_status_trace_log

记录作业状态变更痕迹表可通过每次作业运行的task_id查询作业状态变化的生命轨迹和运行轨迹.

8.2

elastic-job中提供了一个elastic-job-lite-console控制台

设计理念

控制台和Elastic-Job并无直接关系是通过读取Elastic-Job的注册中心数据展示作业状态或更新注册中心数据修改全局配置。

2.控制台只能控制任务本身是否运行但不能控制作业进程的启停因为控制台和作业本身服务器是完全分布式的控制台并不能控制作业服务器。

主要功能:

1.添加作业因为作业都是在首次运行时自动添加使用控制台添加作业并无必要.直接在作业服务器启动包含Elasitc-Job的作业进程即可。

8.2.1

解压缩elastic-job-lite-console-2.1.5.tar

进入bin目录并执行:

bin\start.bat打开浏览器访问http://localhost:8899

用户名:

提供两种用户管理员和访客管理员拥有全部操作权限访客仅拥有查看权限。

默认管理员账号和面膜是root/root,访客用户名和密码是guest/guest,通过conf\auth.properties可以修改管理员以及访客用户名及密码

8.2.2

连接成功后在作业纬度下可以显示该命名空间作业名称分片数量及该作业的cron表达式等信息

在服务器纬度可以查看到服务器ip,当前运行的是实例数作业总数等信息。