【微服务】分布式调度框架PowerJob使用详解

一、前言

二、定时任务调度框架概述

2.1 为什么需要定时任务调度框架

2.2 定时任务调度使用场景

三、PowerJob 介绍

3.1 PowerJob 概述

3.2 PowerJob 功能特性

3.3 PowerJob 应用场景

3.4 PowerJob 与其他同类产品对比

四、PowerJob 部署

4.1 PowerJob 架构

4.2 部署方式介绍

4.3 idea本地部署

4.3.1 获取源码

4.3.2 导入idea

4.3.3 创建数据库

4.3.4 修改配置文件

4.3.5 启动服务

4.3.6 用户注册

4.4 linux部署

4.4.1 本地打包

4.4.2 上传服务器并启动

五、PowerJob快速使用

5.1 新建任务

5.2 修改配置文件并启动服务

5.3 任务执行日志

六、springboot接入PowerJob

6.1 springboot接入PowerJob操作过程

6.1.1 引入依赖包

6.1.2 配置连接信息

6.1.3 自定义执行器类

6.1.4 新建任务并使用自定义执行器

6.1.5 执行任务

七、写在文末

一、前言

定时调度这类需求，在很多项目的业务场景中基本上都会涉及到，尤其是分布式微服务项目，涉及到调度的场景随处可见。对一个运行中的项目来说，接入定时调度技术难度并不大，比如像spring自带的Scheduled 注解，老牌调度框架Quartz ，以及Spring 提供了 TaskScheduler 和 ThreadPoolTaskScheduler 接口，都可用于调度任务的执行。

但是随着项目架构的微服务化改造之后，传统的只有单一调度功能的技术组件很难再满足复杂业务场景下的调度需求，比如任务重试，任务转移等，这就需要功能更强大的带有分布式调度的组件，像xxl-job，ElasticJob，PowerJob 等，不仅功能更强大，而且提供了可视化的UI配置界面，能够更好的适配分布式调度任务下的各种场景。

二、定时任务调度框架概述

2.1 为什么需要定时任务调度框架

定时调度任务框架在实际应用中有非常重要的作用，具体来说：

自动化操作
- 定时调度任务框架能够自动执行预定的操作，如数据备份、系统监控等，减少人为干预的需要，提高效率。
任务管理：
- 通过任务框架可以对各种任务进行管理和监控，包括任务执行状态、时间安排以及错误处理等。
资源优化
- 合理地安排任务的执行时间，可以更好地利用系统资源，避免任务之间的冲突和资源浪费。
提高可靠性
- 通过定时调度任务框架，可以确保重要任务按时执行，从而提高系统的可靠性和稳定性。

2.2 定时任务调度使用场景

定时任务调度框架适用于很多场景，包括但不限于以下几个方面：

数据处理和清洗：
- 定时任务可以用于对数据进行周期性的处理和清洗，例如数据备份、数据同步、数据转换等操作。
定时任务调度：
- 显而易见，定时任务框架最主要的用途就是调度各种类型的定时任务，确保任务按照预定的时间和规则执行。
系统监控和报告：
- 定时任务可用于收集系统指标、生成报告和执行健康检查，帮助监控系统的运行状态并及时发现问题。
发送通知：
- 定时任务可以用于定期发送邮件通知，短信，例如定时发送报表、定时提醒等。
定时数据采集：
- 定时任务可用于周期性地从外部数据源获取数据，用于分析、处理或展示。
缓存刷新：
- 定时任务可用于定期刷新缓存数据，保持缓存数据的最新性，提高系统性能。

三、PowerJob 介绍

3.1 PowerJob 概述

PowerJob 是一个基于分布式任务调度和数据处理框架，用于解决任务调度和数据处理的问题。它使用 Spring Boot 和 Spring Cloud 进行开发，支持多种任务类型，包括定时任务、异步任务、数据处理等。

官网地址：PowerJob

具体来说：

它提供了可靠的任务调度、分布式计算、高可用性和自动扩展等功能，使得分布式任务调度更加方便、高效和可靠；
PowerJob 基于分布式协调服务（如ZooKeeper）和分布式消息队列（如RocketMQ）实现任务的调度和执行；
PowerJob 支持将任务分布在多台机器上执行，通过对任务的监控和管理实现任务的可靠执行；
PowerJob还提供了丰富的任务调度功能，如定时触发、依赖关系、失败重试、超时控制等，满足不同业务场景下的任务调度需求；

3.2 PowerJob 功能特性

PowerJob（原OhMyScheduler）是全新一代分布式任务调度与计算框架，其主要功能特性如下：

使用简单
- 提供前端Web界面，允许开发者可视化地完成调度任务的管理（增、删、改、查）、任务运行状态监控和运行日志查看等功能。
定时策略完善
- 支持 CRON 表达式、固定频率、固定延迟和API四种定时调度策略。
执行模式丰富
- 支持单机、广播、Map、MapReduce 四种执行模式，其中 Map/MapReduce 处理器能使开发者寥寥数行代码便获得集群分布式计算的能力。
工作流支持
- 支持在线配置任务依赖关系（DAG），以可视化的方式对任务进行编排，同时还支持上下游任务间的数据传递，以及多种节点类型（判断节点 & 嵌套工作流节点）。
执行器支持广泛
- 支持 Spring Bean、内置/外置 Java 类，另外可以通过引入官方提供的依赖包，一键集成 Shell、Python、HTTP、SQL 等处理器，应用范围广。
运维便捷
- 支持在线日志功能，执行器产生的日志可以在前端控制台页面实时显示，降低 debug 成本，极大地提高开发效率。
依赖精简
- 最小仅依赖关系型数据库（MySQL/**PostgreSQL/Oracle/MS SQLServer...）**
高可用 & 高性能
- 调度服务器经过精心设计，一改其他调度框架基于数据库锁的策略，实现了无锁化调度。部署多个调度服务器可以同时实现高可用和性能的提升（支持无限的水平扩展）。
故障转移与恢复
- 任务执行失败后，可根据配置的重试策略完成重试，只要执行器集群有足够的计算节点，任务就能顺利完成。

3.3 PowerJob 应用场景

PowerJob 具备其他的任务调度框架的功能同时，还加入了一些其他的新功能特性，比如工作流，分布式处理业务等，官方给出的场景应用如下：

有定时执行需求的业务场景：如每天凌晨全量同步数据、生成业务报表、未支付订单超时取消等；
有需要全部机器一同执行的业务场景：如使用广播执行模式清理集群日志；
有需要分布式处理的业务场景：比如需要更新一大批数据，单机执行耗时非常长，可以使用Map/MapReduce 处理器完成任务的分发，调动整个集群加速计算；
有需要延迟执行某些任务的业务场景：比如订单过期处理等。

更通用来说，PowerJob 具备如下普适的使用场景：

定时任务调度
- PowerJob 提供了灵活的定时任务调度功能，可以按照设定的时间规则执行任务，适用于定时统计、数据备份等场景。
分布式任务协调
- PowerJob 可以协调和管理分布式系统中的各个任务节点，确保任务的顺利执行和结果的一致性。
数据处理和计算
- PowerJob 可用于处理大规模数据集的计算和处理任务，支持并行计算和分布式处理，提高任务执行效率。
实时数据处理
- PowerJob 支持实时数据处理任务，可以满足对数据处理速度要求较高的应用场景，如实时监控、实时报表生成等。
弹性扩展
- PowerJob 可以根据任务负载自动进行弹性扩展，提高系统的稳定性和可靠性。

3.4 PowerJob 与其他同类产品对比

PowerJob 的设计目标为企业级的分布式任务调度平台，即成为公司内部的任务调度中间件。整个公司统一部署调度中心 powerjob-server，旗下所有业务线应用只需要依赖 powerjob-worker 即可接入调度中心获取任务调度与分布式计算能力。

在任务调度的解决方案中，行业内也有不少知名成熟的组件，比如xxl-job，elastic-job等，下面列举几种常用的任务调度技术框架，方便进行学习和对比。

QuartZ	xxl-job	SchedulerX 2.0	PowerJob
定时类型	CRON	CRON	CRON、固定频率、固定延迟、OpenAPI	CRON、固定频率、固定延迟、OpenAPI
任务类型	内置Java	内置Java、GLUE Java、Shell、Python等脚本	内置Java、外置Java（FatJar）、Shell、Python等脚本	内置Java、外置Java（容器）、Shell、Python等脚本
分布式任务	无	静态分片	MapReduce 动态分片	MapReduce 动态分片
在线任务治理	不支持	支持	支持	支持
日志白屏化	不支持	支持	不支持	支持
调度方式及性能	基于数据库锁，有性能瓶颈	基于数据库锁，有性能瓶颈	不详	无锁化设计，性能强劲无上限
报警监控	无	邮件	短信	邮件，提供接口允许开发者扩展
系统依赖	关系型数据库（MySQL、Oracle...）	MySQL	人民币	任意 Spring Data Jpa支持的关系型数据库（MySQL、Oracle...）
DAG 工作流	不支持	不支持	支持	不支持

四、PowerJob 部署

4.1 PowerJob 架构

下图是powerjob官方架构图，从架构图颜色可以看出主体就分了两个大块和一小条：调度中心、执行器、Akka，关于其核心的组件做如下介绍说明：

调度中心 powerjob-server
- PowerJob 的设计目标为企业级的分布式任务调度平台，即成为调度中间件，让任意业务线的应用仅需要依赖 powerjob-worker 即可获取任务调度与分布式计算的能力。因此，PowerJob 的理想部署模式为一个公司统一部署 powerjob-server 集群，各业务线应用直接接入使用。
执行器 powerjob-worker
- 根据以前对定时任务的理解，用过Quartz的话，这里相当于Job这个接口；用过ElasticJob的话，最起码相当于Job接口中的一种，比如SimpleJob接口；用过xxl-job的话，这里也是同理，相当于使用了注解@XxlJob的方法。因此，所有需要执行的任务，mars酱的理解都需要依赖powerjob-worker的。
Akka ActorSystem
- 基于Actor模型设计的，专用于构建高度并发、分布式和弹性的工具包，号称单台机器上高达 200 亿条消息/秒。从架构图来看，PowerJob用来做数据交换传输，这么牛逼的中间协议处理者，看来PowerJob团队一定是想往大了搞的。

4.2 部署方式介绍

PowerJob 提供了多种部署方式，比如以jar包的方式在服务器部署运行，也可以进行docker容器部署。那么部署的是什么呢？实际上就是在上面架构图中介绍的，即powerjob-server。简单理解就是说，powerjob-server就是一个任务调度中心的服务端，通过 powerjob-server，用户可以方便地管理和调度分布式任务，实现任务的高效执行和监控。同时，它支持任务的依赖关系配置、任务失败重试、任务超时处理等功能，能够帮助用户构建稳定可靠的任务调度系统。

可对比理解很多技术中间件的服务端，像mysql，es等，都需要先搭建一个服务端，然后应用才能接入进去执行自身的业务操作。

powerjob部署方式：

本地idea部署
- 即本地直接运行工程，启动服务端；
- 开发或测试的时候可以使用，生产环境不允许；
linux部署
- 将源码下载之后进行打包构建，然后在linux上启动；
docker部署
- 通过docker命令启动powerjob-server的镜像；

4.3 idea本地部署

4.3.1 获取源码

通过git或gitee下载源码到本地；

这里我使用4.3.9的版本；

4.3.2 导入idea

将工程导入idea之后，工程结构如下，部署测试阶段重点关注下面标注的两个模块：

4.3.3 创建数据库

powerjob运行过程中产生的调度数据需要进行持久化存储，因此需要提前创建数据库，在powerjob-server启动时会自动创建数据表。

CREATE DATABASE IF NOT EXISTS `powerjob-daily` DEFAULT CHARSET utf8mb4;

4.3.4 修改配置文件

找到下面这个配置文件

修改位置1，数据库连接：

修改数据库连接地址jdbc-url，指向自己的数据库地址；
修改数据库连接的用户名和密码；

####### 外部数据库配置（需要用户更改为自己的数据库配置） #######
spring.datasource.core.driver-class-name=com.mysql.cj.jdbc.Driver
spring.datasource.core.jdbc-url=jdbc:mysql://IP:3306/powerjob-daily?useUnicode=true&characterEncoding=UTF-8&serverTimezone=Asia/Shanghai
spring.datasource.core.username=root
spring.datasource.core.password=root
spring.datasource.core.maximum-pool-size=20
spring.datasource.core.minimum-idle=5

修改位置2；

可以使用mongodb，只需开启一下即可；
也可以使用mysql，配置连接一下mysql的连接即可；

####### mongoDB配置，非核心依赖，通过配置 oms.mongodb.enable=false 来关闭 #######
#oms.storage.dfs.mongodb.uri=mongodb+srv://zqq:No1Bug2Please3!@cluster0.wie54.gcp.mongodb.net/powerjob_daily?retryWrites=true&w=majority
oms.storage.dfs.mysql_series.driver=com.mysql.cj.jdbc.Driver
oms.storage.dfs.mysql_series.url=jdbc:mysql://IP:3306/powerjob-daily?useUnicode=true&characterEncoding=UTF-8&serverTimezone=Asia/Shanghai
oms.storage.dfs.mysql_series.username=root
oms.storage.dfs.mysql_series.password=root
oms.storage.dfs.mysql_series.auto_create_table=true

修改位置3，邮件设置

####### 邮件配置（不需要邮件报警可以删除以下配置来避免报错） #######
spring.mail.host=smtp.163.com
spring.mail.username=zqq@163.com
spring.mail.password=GOFZPNARMVKCGONV
spring.mail.properties.mail.smtp.auth=true
spring.mail.properties.mail.smtp.starttls.enable=true
spring.mail.properties.mail.smtp.starttls.required=true

其他的配置保持默认即可，用到的时候再配置也可以。

4.3.5 启动服务

powerjob-server-starter模块是一个springboot工程，直接启动即可，启动完成后可以看到相关的数据表初始化出来了；

访问地址：http://127.0.0.1:7700，来到下面的控制台界面；

4.3.6 用户注册

注册一个应用，这个应用可以理解为一个账户，注册完成之后，用这个应用账号和密码登录

登录之后就来到控制台主页

界面上功能的的具体使用，后文将会陆续提到。

4.4 linux部署

4.4.1 本地打包

使用maven命令打出jar包

4.4.2 上传服务器并启动

这里做下模拟，找到打好的jar，然后使用java -jar命令启动即可

五、PowerJob快速使用

5.1 新建任务

在任务管理 —>新建任务，创建一个新的任务；

填写表单参数，注意标注的是非必填的内容，如果不清楚参数含义的话；

在上面的表单项中，任务每隔10秒执行一次，创建完成之后，在任务列表页就可以看到这条新建的任务。

5.2 修改配置文件并启动服务

找到工程中的worker-samples模块，该模块提供了一些测试用的任务样例，即任务的处理器示例，可直接使用；

使用之前，首先要修改配置文件中的应用名称，使用上午中创建的那个应用；

我们选择样例中的StandaloneProcessorDemo这个类作为测试，在上面的任务表单参数中，也使用的是这个类的全类名，以上配置都完成之后，运行SampleApplication即可。然后在任务列表中点击运行。

稍等一会，在控制台上面就能看到类中输出的日志信息，而后按照没10秒的频率输出，同时还将任务中的参数输出了出来。

5.3 任务执行日志

点击任务后的更多—>运行记录，可以查看任务运行时的日志信息。

点击每一条日志记录进去，可以查看每一条执行日志的详情，如果任务执行过程中有报错也可以通过检查日志进行分析定位。

六、springboot接入PowerJob

如何基于当前的springboot项目接入PowerJob呢？下面看具体的操作步骤。

6.1 springboot接入PowerJob操作过程

6.1.1 引入依赖包

依赖包的版本可以根据自身的需求选择，这里选择版本时，尽量与powerjob服务端版本保持一致

<dependency>
    <groupId>tech.powerjob</groupId>
    <artifactId>powerjob-worker-spring-boot-starter</artifactId>
    <version>4.3.8</version>
</dependency>

6.1.2 配置连接信息

注意应用名称填写上面注册的那个应用名称，这里推荐直接将源码中样例的配置文件拿过来稍做修改即可；

server.port=7788
spring.jpa.open-in-view=false
powerjob.worker.enabled=true
powerjob.worker.allow-lazy-connect-server=false
powerjob.worker.port=27777
powerjob.worker.app-name=app_test
powerjob.worker.server-address=127.0.0.1:7700,127.0.0.1:7701
powerjob.worker.protocol=http
powerjob.worker.store-strategy=disk
powerjob.worker.max-result-length=4096
powerjob.worker.max-appended-wf-context-length=4096

6.1.3 自定义执行器类

自定义一个执行器，实现BasicProcessor接口，然后重写里面的process方法，可以参考上面的源码工程中的示例类变现；

import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import tech.powerjob.worker.core.processor.ProcessResult;
import tech.powerjob.worker.core.processor.TaskContext;
import tech.powerjob.worker.core.processor.sdk.BasicProcessor;

@Component
@Slf4j
public class MyProcessor implements BasicProcessor {

    @Override
    public ProcessResult process(TaskContext taskContext) throws Exception {
        log.info("MyProcessor fetch ...");
        return new ProcessResult(true, taskContext + ": ProcessResult fetch success" );
    }
}