分布式架构与XXL-JOB

先了解什么是任务调度？

什么是分布式任务调度？

了解XXL-JOB分布式任务调度平台

如何搭建XXL-JOB？

分片广播

作业分片方案

最近学习在项目的媒资管理模块如何高效处理大量视频，上传单个视频可能涉及到转码，内容审核等多个处理部分

如果采用传统的单线程处理模式可能会导致资源利用率低下，所有操作依次排队会导致执行处理视频耗时较长，效率不高的问题出现

因此，在处理大量视频文件时，我们需要学习更高效的任务处理方式，如何高效满足这些视频处理需求？

1.采用多线程

在处理大量视频任务时，单线程处理效率较低，无法充分利用系统资源。为了提高处理效率可以采用多线程技术来并发处理任务

核心思想是：将一个任务拆分为多个子任务，并发执行这些子任务，充分利用CPU和内存资源，提高任务处理效率

2.采用分布式架构+多线程

通过分布式架构和多线程技术的结合，最大化利用计算资源，高效处理大规模任务，例如我们所说的处理大量视频文件

采用分布式+线程可扩展性更强，同时它也是一种分布式任务调度的处理方案

先了解什么是任务调度？

任务调度：顾名思义就是对任务的调度，它是指系统为了完成特定的业务，基于给定的时间点，给定的时间间隔或者给定执行次数自动执行任务

我们思考一下下面业务场景

某财务系统需要在每天上午10点结算前一天的账单数据，统计汇总
12306网站会根据车次不同，设置几个时间点分批放票
商品成功发货后，需要向客户发送短信提醒

以上这些场景，就是任务调度所需要解决的问题，类似场景还有很多，我们该如何实现？这里我举例三种实现方法

1.多线程方式实现

回顾多线程思想，当我们可以开启一个线程，每sleep一段时间，就去检查是否已经到预期执行时间

简单实现任务调度的功能（按一定的间隔时间执行任务调度的功能）：

public static void main(String[] args){//任务执行时间间隔final long timeInterval = 1000;Runnable runnable = new Runnable()}{public void run(){while(true){//TODO:somethingtry{Thread.sleep(timeInterval);}catch(InterruptedException e){e.printStackTrace(); } }}};Thread thread = new Thread(runnable); thread.start();  }
}

2.Timer 和 ScheduledExecutor（JDK提供的相关支持）是用于定时任务调度的工具类，帮助我们在指定的时间点或周期性的执行任务

Timer的核心功能：

单次任务：在指定的延迟后执行一次任务
周期性任务：以固定的时间间隔重复执行任务

Timer的优点在于简单易用，每个Timer对应一个线程，因此可以同时启动多个Timer并行执行多个任务，同一个Timer中的任务是串行执行

Timer方法的实现：

public static void main(String[] args){Timer timer = new Timer();timer.schedule(new TimerTask(){@Overridepublic void run(){//TODO:someting  }},1000,2000); //一秒开始调度，每2秒执行一次}

ScheduledExecutor方式实现：

public static void main(String [] args){//创建一个固定大小为10的线程池，用于执行定时任务（线程池中的线程可以并发执行多个任务）ScheduledExecutorService service = Executors.newScheduledThreadPool(10);service.scheduleAtFixedRate(new Runnable(){@Override//任务逻辑是打印“todo something”public void run(){//TODO:somethingSystem.out.print("todo something");}//任务将在一秒后首次执行，之后，任务以每2秒执行一次},1,2,TimeUnit.SECONDS);
}

scheduleAtFixedRate:

是用于以固定的速率调度任务，任务会在指定的初始延迟后开始执行，之后以固定的时间间隔重复执行

基于线程池设计的ScheduledExecutor，其设计思想是，每一个被调度的任务都会由线程池中一个线程去执行，因为任务是并发执行的，相互之间不会受到打扰

Timer和ScheduledExecutor都仅能提供基于开始时间与重复间隔的任务调度，不能胜任更加复杂的调度需求 — 比如，设置每月第一天凌晨1点执行任务，复杂调度任务的管理，任务间传递数据等等

3.再学习另一个功能强大的任务调度框架Quartz，它可以满足更多更复杂的调度需求，Quartz设计的核心类包括Scheduler，Job以及Trigger，其中，Job负责定义需要执行的任务，Trigger负责设置调度策略，Scheduler将两者组装起来，并触发任务开始执行，Quartz支持简单的按时间间隔调度，还支持按日历调度方式，通过设置CronTrigger表达式（包括：秒，分，时，日，月，周，年）进行任务调度

public static void main(String [] agrs) throws SchedulerException {//创建一个SchedulerSchedulerFactory schedulerFactory = new StdSchedulerFactory();Scheduler scheduler = schedulerFactory.getScheduler();//创建JobDetailJobBuilder jobDetailBuilder = JobBuilder.newJob(MyJob.class);jobDetailBuilder.withIdentity("jobName","jobGroupName");JobDetail jobDetail = jobDetailBuilder.build();//创建触发的CronTrigger 支持按日历调度CronTrigger trigger = TriggerBuilder.newTrigger().withIdentity("triggerName", "triggerGroupName").startNow().withSchedule(CronScheduleBuilder.cronSchedule("0/2 * * * * ?")).build();scheduler.scheduleJob(jobDetail,trigger);scheduler.start();
}public class MyJob implements Job {@Overridepublic void execute(JobExecutionContext jobExecutionContext){System.out.println("todo something");}
}

通过以上内容学习了什么是任务调度，任务调度所解决的问题，以及任务调度的多种实现方式

什么是分布式任务调度？

通常任务调度的程序是集成在应用中的，比如：优惠卷服务中包括了定时发放优惠卷的的调度程序，结算服务中包括了定期生成报表的任务调度程序，由于采用分布式架构，一个服务往往会部署多个冗余实例来运行我们的业务，在这种分布式系统环境下运行任务调度，我们称之为分布式任务调度，如下图：

分布式调度要实现的目标：

不管是任务调度程序集成在应用程序中，还是单独构建的任务调度系统，如果采用分布式调度任务的方式就相当于将任务调度程序分布式构建，这样就可以具有分布式系统的特点，并且提高任务的调度处理能力

1.并行任务调度

并行任务调度实现靠多线程，如果有大量任务需要调度，此时光靠多线程就会有瓶颈了，因为一台计算机CPU的处理能力是有限的

如果将任务调度程序分布式部署，每个结点还可以部署为集群，这样就可以让多台计算机共同去完成任务调度，我们可以将任务分割为若干个分片，由不同的实例并行执行，来提高任务调度的处理效率。

2、高可用

若某一个实例宕机，不影响其他实例来执行任务。

3、弹性扩容

当集群中增加实例就可以提高并执行任务的处理效率。

4、任务管理与监测

对系统中存在的所有定时任务进行统一的管理及监测。让开发人员及运维人员能够时刻了解任务执行情况，从而做出快速的应急处理响应。

5、避免任务重复执行

当任务调度以集群方式部署，同一个任务调度可能会执行多次，比如在上面提到的电商系统中到点发优惠券的例子，就会发放多次优惠券，对公司造成很多损失，所以我们需要控制相同的任务在多个运行实例上只执行一次。

了解XXL-JOB分布式任务调度平台

XXL-JOB是一个 轻量级分布式任务调度平台，专门解决分布式系统中的定时任务调度问题，其核心设计目标是开发迅速、学习简单、轻量级、易扩展

官网：https://www.xuxueli.com/xxl-job/

XXL-JOB（如图所示）主要有调度中心，执行器，任务：

1.调度中心（XXL-JOB Admin）:

调度中心是XXL-JOB的核心管理平台，负责对任务进行统一的管理和调度，它提供了一个可视化的Web界面，方便开发者进行任务的配置，监控和管理

2.执行器（XXL-JOB Executor）

执行器是实际执行任务的组件，它负责接收调度中心发送的任务请求，并执行具体的业务逻辑，执行器可以部署在多个节点上，实现分布式任务执行

3.任务

负责执行具体的业务逻辑

调度中心与执行器之间的工作流程如下：

执行流程：

1.任务执行器根据配置的调度中心的地址，自动注册到调度中心

2.达到任务触发条件，调度中心下发任务

3.执行器基于线程池执行任务，并将执行结果放到内存队列中，把执行日志写入日志文件中

4.执行器消费内存队列中的执行结果，主动上报给调度中心

5.当用户在调度中心查看任务日志，调度中心请求任务执行器，任务执行器读取任务日志文件并返回日志详情

如何搭建XXL-JOB？

首先需要下载XXL-JOB，可以从GitHub地址或者Gitee地址进行下载

GitHub 地址：

仓库地址：GitHub - xuxueli/xxl-job: A distributed task scheduling framework.（分布式任务调度平台XXL-JOB）

Gitee 地址：

仓库地址：xxl-job: 一个分布式任务调度平台，其核心设计目标是开发迅速、学习简单、轻量级、易扩展。现已开放源代码并接入多家公司线上产品线，开箱即用。

如何集成XXL-JOB？我在我的媒资管理模块项目中进行演示

1.当下载好后，用IDEA打开目录显示界面如下

其中的 xxl-job-admin为调度中心，xxl-job-core为公共依赖，xxl-job-executor-samples为执行器（推荐使用Springboot版本，通过Springboot管理执行器）

2.再创建一个命名为xxl_job_2.3.1的数据库

CREATE DATABASE xxl_job_2.3.1 DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci;

根据数据库脚本创建数据库，修改数据库连接信息和端口，启动xxl-job-admin，访问http://local:18088/xxl-job-admin/（分布式任务调度平台 — 管理和调度分布式环境中的定时任务）

使用用户名：admin / 密码：123456 登录

下面配置执行器，执行器负责与调度中心通信接收调度中心发起的任务调度请求

1.进入调度中心添加执行器

启动成功之后，可以选择在Linux上运行

使用maven命令，将xxl-job-admin打包，然后将其上传至Linux中，使用命令启动

nohup java -jar /绝对路径/xxl-job-admin-2.3.1.jar &

添加完执行器后，接着配置执行器，执行器负责与调度中心通信，接收调度中心发起的任务调度请求

1.首先在media-service工程中添加依赖（父工程中完成了版本控制，这里的版本是2.3.1）

<dependency><groupId>com.xuxueli</groupId><artifactId>xxl-job-core</artifactId>
</dependency>

2.在nacos下的media-service-dev.yaml下配置xxl-job

注意这里配置的appname是执行器的应用名，稍后会在调度中心配置执行器的时候使用

xxl:job:admin:addresses: http://192.168.101.128:18088/xxl-job-admin/executor:appname: media-process-serviceaddress:ip:port: 9999logpath: /data/applogs/xxl-job-jobhandlerlogretentiondays: 30accessToken: default_token

3.配置xxl-job的执行器

将xxl-job-executor-sample-springboot示例工程下的配置类拷贝到媒资管理的service工程下：

该类中的属性就是获取配置文件中的配置得到的，同时提供了一个执行器的Bean，用于初始化XXL-JOB执行器的核心组件XxIJobSpringExecutor

并将其注册为Spring Bean，使得执行器能够连接到XXL-JOB管理平台并接收任务调度请求

@Configuration
public class XxlJobConfig {private Logger logger = LoggerFactory.getLogger(XxlJobConfig.class);@Value("${xxl.job.admin.addresses}")private String adminAddresses;@Value("${xxl.job.accessToken}")private String accessToken;@Value("${xxl.job.executor.appname}")private String appname;@Value("${xxl.job.executor.address}")private String address;@Value("${xxl.job.executor.ip}")private String ip;@Value("${xxl.job.executor.port}")private int port;@Value("${xxl.job.executor.logpath}")private String logPath;@Value("${xxl.job.executor.logretentiondays}")private int logRetentionDays;@Beanpublic XxlJobSpringExecutor xxlJobExecutor() {logger.info(">>>>>>>>>>> xxl-job config init.");XxlJobSpringExecutor xxlJobSpringExecutor = new XxlJobSpringExecutor();xxlJobSpringExecutor.setAdminAddresses(adminAddresses);xxlJobSpringExecutor.setAppname(appname);xxlJobSpringExecutor.setAddress(address);xxlJobSpringExecutor.setIp(ip);xxlJobSpringExecutor.setPort(port);xxlJobSpringExecutor.setAccessToken(accessToken);xxlJobSpringExecutor.setLogPath(logPath);xxlJobSpringExecutor.setLogRetentionDays(logRetentionDays);return xxlJobSpringExecutor;}/*** 针对多网卡、容器内部署等情况，可借助 "spring-cloud-commons" 提供的 "InetUtils" 组件灵活定制注册IP；**      1、引入依赖：*          <dependency>*             <groupId>org.springframework.cloud</groupId>*             <artifactId>spring-cloud-commons</artifactId>*             <version>${version}</version>*         </dependency>**      2、配置文件，或者容器启动变量*          spring.cloud.inetutils.preferred-networks: 'xxx.xxx.xxx.'**      3、获取IP*          String ip_ = inetUtils.findFirstNonLoopbackHostInfo().getIpAddress();*/}

再次进入调度中心，添加执行器

重启媒资管理服务模块，就可以看到执行器在调度中心注册成功

下面我们编写任务演示如何使用，在media-service下新建包com.project.media.service.jobhandler,在该包下定义我们的任务类

package com.xuecheng.media.service.jobhandler;import com.xxl.job.core.context.XxlJobHelper;
import com.xxl.job.core.handler.annotation.XxlJob;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;import java.util.concurrent.TimeUnit;/*** @description 测试执行器* @author Mr.M* @date 2022/9/13 20:32* @version 1.0*/@Component@Slf4j
public class SampleJob {/*** 1、简单任务示例（Bean模式）*/@XxlJob("testJob")public void testJob() throws Exception {log.info("开始执行.....");}}

然后进入调度中心添加任务，进入任务管理，新增任务信息

调度类型：固定速度指按固定的间隔定时调度

Cron：通过Cron表达式实现更丰富的定时调度策略，它的表达式是一个字符串，通过它可以定义调度策略，格式如下：

{秒数}{分数}{小时}{日期}{月份}{星期}{年份（可为空）}

比如：

30 10 1**？表示每天1点10分30秒触发

0/30****？每30秒触发一次

* 0/10***？每10分钟触发一次

添加成功，启动任务

下面启动媒资管理的service工程，启动执行器并观察执行器方法的执行

掌握了xxL-Job的基本使用，继续思考如何进行分布式任务处理？比如说我们会启动多个执行器组成一个集群，去执行任务

结合以上XXL-JOB的实现，我们可以利用 分片广播策略 来提高处理效率尤其是处理数据库中大量数据（比如项目中媒资管理模块需要处理大量视频文件）

分片广播

分片广播是分布式任务调度中一种常见模式，主要用于将一个大任务拆分成多个子任务（分片），并将这些子任务 并发分发 到多个执行器节点上执行，从而提高任务处理效率

以上图示是调度中心以执行器为维度进行分片，将集群中的执行器标上序号：0，1，2，3....,广播是指每次调度会向集群中的所有执行器发送任务调度，请求中携带分片参数，每个执行器收到调度请求同时接收分片参数

像这种分片广播适用于哪些场景？

分片任务场景：10个执行器的集群来处理10w条数据（数据之多），每台机器只需要处理1w条数据（分片执行），耗时降低10倍;

广播任务场景：广播执行器同时运行shell脚本，广播集群节点进行缓存更新等

"分片广播" 和普通任务开发流程一致，不同之处在于可以获取分片参数进行分片业务处理。

测试分片广播策略

1.定义作业分片的任务方法

/***分片广播任务*/@XxlJob("shardingJobHandler")public void shardingJobHandler() throws Exception{//分片参数int shardIndex = XxlJobHelper.getShardIndex();int shardTotal = XxlJobHelper.getShardTotal();Log.info("分片参数：当前分片序号 = {}，总分片数 = {}"，shardIndex，shardTotal);
}

2.在调度中心添加任务

回到任务调度界面，可以看到该项任务添加成功

3.启动任务，观察日志

下面需要启动两个执行器实例，观察每个实例的执行情况

1.首先在nacos中配置media-service的本地优先部署

#配置本地优先
spring:cloud:config:override-none: true

2.将media-service启动两个实例，两个实例在启动时端口不能冲突

例如：在VM options处添加：-Dserver.port=63051 - Dxxl.job.executor.port=9998

在VM options处添加： -Dserver.port=63050 - Dxxl.job.executor.port=9999

3.启动两个实例，观察任务调度中心，稍等片刻执行器会有两个

分别观察这两个执行实例的日志

日志一：

日志二：

从日志可以看到每个实例的分片序号不同，如果其中一个执行器挂掉，只剩下一个执行器在工作，稍等片刻调用中心发现少了一个执行器将动态调整总分片数为1

作业分片方案

以上任务添加成功后，对于要处理的任务会添加到待处理任务表中，现在启动多个执行器实例去查询这些待处理任务，如何保证多个执行器不会查询到重复的任务呢？

XXL-JOB并不直接提供数据处理的功能，它只会给执行器分配好分片序号，在向执行器任务调度的同时下发分片总数以及分片序号等这些参数，执行器收到这些参数后根据自己的业务需求去利用这些参数

下图表示了多个执行器获取视频处理任务的结构：

每个执行器收到广播任务会有两个参数：分片总数，分片序号

每次执行从数据表取任务时可以让任务id 模上（取模运算）分片总数，如果等于分片序号则能执行此任务

假设我们有6个待处理的视频任务，任务ID依次为1-6，同时有3个执行器实例，那么分片总数就是3，对应的分片序号则为0，1，2。具体任务分配情况如下：

任务序号	任务 ID 分片总数（求模运算）	应执行的执行器分片序号
1	1 % 3 = 1	执行器 2
2	2 % 3 = 2	执行器 3
3	3 % 3 = 0	执行器 1
4	4 % 3 = 1	执行器 2
5	5 % 3 = 2	执行器 3
6	6 % 3 = 0	执行器 1

从上述表格中可以清晰地看到，每个任务都根据其任务ID的取模结果被精准地分配到了对应的执行器上，而不会出现多个执行器获取到同一个任务的情况

通过这种XXL-JOB的分片机制，可以用于处理视频转码，视频审核，视频分发等任务，将大量视频处理任务均匀分配给多个执行器，保证任务的高效执行和不会重复处理