分布式任务调度框架使用的需求背景和痛点和探索

手赚控 2023年08月29日 05:10 234 0

分布式任务调度框架几乎是每位小型应用必备的工具，本文介绍了任务调度框架使用的需求背景和痛点，对业界普遍使用的开源分布式任务调度框架的使用进行了探究实践，并剖析了这几种框架的优缺点和对自身业务的思索。

一、业务背景

1.1为何还要使用定时任务调度

（1）时间驱动处理场景：整点发送降价券，每次更新回报，每次刷新标签数据和人群数据。

（2）批量处理数据：足额批量统计报表数据，批量更新邮件状态，实时性要求不高。

（3）异步执行时延：活动状态刷新，异步执行离线查询，与内部逻辑时延。

1.2使用需求和痛点

（1）任务执行监控信令能力。

（2）任务可灵活动态配置，无需重启。

（3）业务透明，低耦合，配置精简，开发便于。

（4）易检测。

（5）高可用，无单点故障。

（6）任务不可重复执行，避免逻辑异常。

（7）大任务的分发并行处理能力。

二、开源框架实践与探求

2.1Java原生Timer和

vice

2.1.1Timer使用

Timer缺陷：

Timer底层是使用单句柄来处理多个Timer任务，这意味着所有任务实际上都是串口执行，前一个任务的推迟会影响到后来的任务的执行。

因为单句柄的缘故，即便某个定时任务在运行时，形成未处理的异常，这么除了当前这个句柄会停止，所有的定时任务就会停止。

Timer任务执行是依赖于系统绝对时间，系统时间变化会造成执行计划的变更。

因为上述缺陷，尽量不要使用Timer，idea中也会明晰提示，使用代替Timer。

2.1.2vice使用

vice对于Timer的缺陷进行了修复，首先vice内部实现是轮询池，可以支持多个任务并发执行。

对于某一个句柄执行的任务出现异常，也会处理，不会影响其他句柄任务的执行，另外vice是基于时间间隔的推迟，执行不会因为系统时间的改变发生变化。

其实，vice还有自己的局限性：只好依照任务的推迟来进行调度，难以满足基于绝对时间和月历调度的需求。

2.2Task

2.2.1Task使用

task是自主开发的羽量级定时任务框架，不须要依赖其他额外的包，配置较为简略。

此处使用注解配置

2.2.2Task缺陷

Task原本不支持持久化，也没有推出官方的分布式集群方式，只好靠开发者在业务应用中自己自动扩充实现，难以满足可视化，易配置的需求。

2.3永远精典的

2.3.1基本介绍

框架是Java领域最知名的开源任务调度工具，只是现在事实上的定时任务标准，几乎全部的开源定时任务框架都是基于核心调度建立而成。

2.3.2原理解读

核心组件和构架

关键概念

（1）：任务调度器，是执行任务调度的控制器。本质上是一个计划调度容器，注册了全部和对应的，使用句柄池作为任务运行的基础组件任务平台，增加任务执行效率。

（2）：触发器，适于定义任务调度的时间规则，告诉任务调度器哪些时侯触发任务，其中是基于cron式子建立的功能强悍的触发器。

（3）：月历特定时间点的集合。一个可以包含多个，可适于排除或包含这些时间点。

（4）：是一个可执行的工作，拿来描述Job实现类及其它相关的静态信息，如Job的名称、监听器等相关信息。

（5）Job：任务执行插口，只有一个办法，适于执行真正的业务逻辑。

（6）：任务储存方法，主要有和，是储存在JVM的显存中，有遗失和数目受限的风险，是将任务信息持久化到数据库中，支持集群。

2.3.3实践说明

（1）关于的基本使用

（2）业务使用要满足动态更改和重启不遗失，通常还要使用数据库进行保存。

（3）组件化

（4）扩充

2.3.4缺陷和不足

（1）还要把任务信息持久化到业务数据表，和业务有耦合。

（2）调度逻辑和执行逻辑并存于同一个项目中，在机器功耗固定的状况下，业务和调度之间不可防止地会互相影响。

（3）集群方式下，是通过数据库独霸锁来惟一获取任务，任务执行并没有实现建立的负载均衡体系。

2.4羽量级利器XXL-JOB

2.4.1基本介绍

XXL-JOB是一个羽量级分布式任务调度平台，主打特征是平台化，易布署，开发快速、学习简略、轻量级、易扩充，代码仍在持续更新中。

“调度中心”是任务调度控制台，平台自身并不承当业务逻辑，也是负责任务的统一管理和调度执行，使得提供任务管理平台，“执行器”负责接收“调度中心”的调度并执行，可直接布署执行器，也可以将执行器集成到现有业务项目中。通过将任务的调度控制和任务的执行时延，业务使用只须要关注业务逻辑的开发。

主要提供了任务的动态配置管理、任务监控和统计报表以及调度日志几大功能模块，支持多种运行方式和路由策略，可基于对应执行器机器集群人数进行简略分片数据处理。

2.4.2原理解读

2.1.0版本前核心调度模块都是基于框架，2.1.0版本开始自研调度组件，移除依赖，使用时间轮调度。

2.4.3实践说明

具体配置和介绍参考官方文档。

2.4.3.1demo使用：

样例1：实现简略任务配置，只须要承继具象类，并申明注解

@(value="r")，实现业务逻辑即可。（注：这次引进了dubbo，后文介绍）。

@JobHandler(value="offlineTaskJobHandler")@Componentpublic class OfflineTaskJobHandler extends IJobHandler {  @Reference(check = false,version = "cms-dev",group="cms-service") private OfflineTaskExecutorFacade offlineTaskExecutorFacade;  @Override public ReturnT execute(String param) throws Exception { XxlJobLogger.log(" offlineTaskJobHandler start.");  try { offlineTaskExecutorFacade.executeOfflineTask(); } catch (Exception e) { XxlJobLogger.log("offlineTaskJobHandler-->exception." , e); return FAIL; }  XxlJobLogger.log("XXL-JOB, offlineTaskJobHandler end."); return SUCCESS; }}

（滑动可查看）

样例2：分片广播任务。

@JobHandler(value="shardingJobHandler")@Servicepublic class ShardingJobHandler extends IJobHandler {  @Override public ReturnT execute(String param) throws Exception {  // 分片参数 ShardingUtil.ShardingVO shardingVO = ShardingUtil.getShardingVo(); XxlJobLogger.log("分片参数：当前分片序号 = {}, 总分片数 = {}", shardingVO.getIndex(), shardingVO.getTotal());  // 业务逻辑 for (int i = 0; i < shardingVO.getTotal(); i++) { if (i == shardingVO.getIndex()) { XxlJobLogger.log("第 {} 片, 命中分片开始处理", i); } else { XxlJobLogger.log("第 {} 片, 忽略", i); } }  return SUCCESS; }}

（滑动可查看）

2.4.3.2整合dubbo

（1）引进dubbo--boot-和业务jar包依赖。

<dependency> <groupId>com.alibaba.spring.bootgroupId> <artifactId>dubbo-spring-boot-starterartifactId> <version>2.0.0version>dependency> <dependency> <groupId>com.demo.servicegroupId> <artifactId>xxx-facadeartifactId> <version>1.9-SNAPSHOTversion>dependency>

（滑动可查看）

（2）配置文件加入dubbo消费端配置（可依据环境定义多个配置文件，通过切换）。

## Dubbo 服务消费者配置spring.dubbo.application.name=xxl-job spring.dubbo.registry.address=zookeeper://zookeeper.xyz:2183spring.dubbo.port=20880 spring.dubbo.version=demospring.dubbo.group=demo-service

（滑动可查看）

（3）代码中通过@注入插口即可。

@Reference(check = false,version = "demo",group="demo-service")private OfflineTaskExecutorFacade offlineTaskExecutorFacade;

（滑动可查看）

（4）启动程序加入@tion注解。


@SpringBootApplication@EnableDubboConfigurationpublic class XxlJobExecutorApplication { public static void main(String[] args) { SpringApplication.run(XxlJobExecutorApplication.class, args); }}

（滑动可查看）

2.4.4任务可视化配置

外置了平台项目，便于了开发者对任务的管理和执行日志的监控，并提供了一些方便检测的功能。

2.4.5扩充

（1）任务监控和报表的优化。

（2）任务报案形式的扩充，例如加入信令中心，提供内部消息，邮件信令。

（3）对实际业务内部执行出现异常状况下的不同监控信令和重试策略。

2.5高可用-Job

2.5.1基本介绍

-Job是一个分布式调度解决方案，由两个互相独立的子项目-Job-Lite和-Job-Cloud组成。

-Job-Lite定位为羽量级无中心化解决方案，使用jar包的方式提供分布式任务的协调服务。

-Job-Cloud使用Mesos+的解决方案，额外提供资源整治、应用分发以及进程隔离等服务。

可惜的是早已三年没有迭代更新记录。

2.5.2原理解读

2.5.3实践说明

2.5.3.1demo使用

（1）安装，配置注册中心，配置文件加入注册中心zk的配置。

@Configuration@ConditionalOnExpression("'${regCenter.serverList}'.length() > 0")public class JobRegistryCenterConfig {  @Bean(initMethod = "init") public ZookeeperRegistryCenter regCenter(@Value("${regCenter.serverList}") final String serverList, @Value("${regCenter.namespace}") final String namespace) { return new ZookeeperRegistryCenter(new ZookeeperConfiguration(serverList, namespace)); }}

放单任务平台_任务平台_任务平台发布任务赚钱

（滑动可查看）

spring.application.name=demo_elasticjob regCenter.serverList=localhost:2181regCenter.namespace=demo_elasticjob spring.datasource.url=jdbc:mysql://127.0.0.1:3306/xxl-job?Unicode=true&characterEncoding=UTF-8spring.datasource.username=userspring.datasource.password=pwd

（滑动可查看）

（2）配置数据源，并配置文件中加入数据源配置。

@Getter@Setter@NoArgsConstructor@AllArgsConstructor@ToString@Configuration@ConfigurationProperties(prefix = "spring.datasource")public class DataSourceProperties { private String url; private String username; private String password;  @Bean @Primary public DataSource getDataSource() { DruidDataSource dataSource = new DruidDataSource(); dataSource.setUrl(url); dataSource.setUsername(username); dataSource.setPassword(password); return dataSource; }}

（滑动可查看）

spring.datasource.url=jdbc:mysql://127.0.0.1:3306/xxl-job?Unicode=true&characterEncoding=UTF-8spring.datasource.username=userspring.datasource.password=pwd

（滑动可查看）

（3）配置风波。

@Configurationpublic class JobEventConfig { @Autowired private DataSource dataSource;  @Bean public JobEventConfiguration jobEventConfiguration() { return new JobEventRdbConfiguration(dataSource); }}

（滑动可查看）

（4）为了方便灵活配置不同的任务触发风波，加入注解。

@Target({ElementType.TYPE})@Retention(RetentionPolicy.RUNTIME)public @interface ElasticSimpleJob {  @AliasFor("cron") String value() default "";  @AliasFor("value") String cron() default "";  String jobName() default "";  int shardingTotalCount() default 1;  String shardingItemParameters() default "";  String jobParameter() default "";}

（滑动可查看）

（5）对配置进行初始化。

@Configuration@ConditionalOnExpression("'${elaticjob.zookeeper.server-lists}'.length() > 0")public class ElasticJobAutoConfiguration {  @Value("${regCenter.serverList}") private String serverList;  @Value("${regCenter.namespace}") private String namespace;  @Autowired private ApplicationContext applicationContext; @Autowired private DataSource dataSource;  @PostConstruct public void initElasticJob() { ZookeeperRegistryCenter regCenter = new ZookeeperRegistryCenter(new ZookeeperConfiguration(serverList, namespace)); regCenter.init(); Map<String, SimpleJob> map = applicationContext.getBeansOfType(SimpleJob.class);  for (Map.Entry<String, SimpleJob> entry : map.entrySet()) { SimpleJob simpleJob = entry.getValue(); ElasticSimpleJob elasticSimpleJobAnnotation = simpleJob.getClass().getAnnotation(ElasticSimpleJob.class);  String cron = StringUtils.defaultIfBlank(elasticSimpleJobAnnotation.cron(), elasticSimpleJobAnnotation.value()); SimpleJobConfiguration simpleJobConfiguration = new SimpleJobConfiguration(JobCoreConfiguration.newBuilder(simpleJob.getClass().getName(), cron, elasticSimpleJobAnnotation.shardingTotalCount()).shardingItemParameters(elasticSimpleJobAnnotation.shardingItemParameters()).build(), simpleJob.getClass().getCanonicalName()); LiteJobConfiguration liteJobConfiguration = LiteJobConfiguration.newBuilder(simpleJobConfiguration).overwrite(true).build();  JobEventRdbConfiguration jobEventRdbConfiguration = new JobEventRdbConfiguration(dataSource); SpringJobScheduler jobScheduler = new SpringJobScheduler(simpleJob, regCenter, liteJobConfiguration, jobEventRdbConfiguration); jobScheduler.init(); } }}

（滑动可查看）

（6）实现插口任务平台，按上文中方式整合dubbo,完成业务逻辑。

@ElasticSimpleJob( cron = "*/10 * * * * ?", jobName = "OfflineTaskJob", shardingTotalCount = 2, jobParameter = "测试参数", shardingItemParameters = "0=A,1=B")@Componentpublic class MySimpleJob implements SimpleJob { Logger logger = LoggerFactory.getLogger(OfflineTaskJob.class);  @Reference(check = false, version = "cms-dev", group = "cms-service") private OfflineTaskExecutorFacade offlineTaskExecutorFacade;   @Override public void execute(ShardingContext shardingContext) {  offlineTaskExecutorFacade.executeOfflineTask();  logger.info(String.format("Thread ID: %s, 作业分片总数: %s, " + "当前分片项: %s.当前参数: %s," + "作业名称: %s.作业自定义参数: %s" , Thread.currentThread().getId(), shardingContext.getShardingTotalCount(), shardingContext.getShardingItem(), shardingContext.getShardingParameter(), shardingContext.getJobName(), shardingContext.getJobParameter() )); }}