PowerJob 分布式任务调度实战：可视化控制台 + 容器化部署 + 故障转移

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

┌────────────────────────────────────────┐
│           PowerJob Server              │
│   (调度中心，端口 7700 HTTP / 10010 OMS)│
│  ┌──────────┐ ┌──────────┐ ┌────────┐ │
│  │  调度器  │ │  执行器  │ │ 控制台 │ │
│  └─────┬────┘ └─────┬────┘ └────┬───┘ │
└────────┼────────────┼──────────┼──────┘
         │            │          │
         ↓            ↓          ↓
┌────────────────────────────────────────┐
│            Worker 集群                 │
│  ┌──────┐  ┌──────┐  ┌──────┐         │
│  │Worker│  │Worker│  │Worker│         │
│  │  1   │  │  2   │  │  3   │         │
│  └──────┘  └──────┘  └──────┘         │
└────────────────────────────────────────┘
         ↓
      MySQL（持久化）

1
2
3
4
5
6
7

# 1. 创建数据库
mysql -u root -p{{MYSQL_PASSWORD}} -e \
  "CREATE DATABASE powerjob CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;"

# 2. 初始化表结构
# https://github.com/PowerJob/PowerJob/blob/v4.3.9/others/powerjob-mysql.sql
mysql -u root -p{{MYSQL_PASSWORD}} powerjob < powerjob-mysql.sql

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

# docker-compose.yml
version: '3'
services:
  powerjob-server:
    image: powerjob/powerjob-server:V4.3.9
    container_name: powerjob
    restart: always
    network_mode: host
    environment:
      - TZ=Asia/Shanghai
      - PARAMS=--spring.profiles.active=product \
               --spring.datasource.core.jdbc-url=jdbc:mysql://{{MYSQL_HOST}}:3306/powerjob?useUnicode=true&characterEncoding=UTF-8 \
               --spring.datasource.core.username={{DB_USER}} \
               --spring.datasource.core.password={{DB_PASSWORD}} \
               --oms.http.port=10010 \
               --server.address={{SERVER_IP}}
    volumes:
      - /data/powerjob-server:/root/powerjob/server
      - ~/.m2:/root/.m2

1

docker-compose up -d

1

http://{{SERVER_IP}}:7700

1
2
3
4
5

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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

powerjob:
  worker:
    enabled: true
    allow-lazy-connect-server: false   # 启动时立即连 Server
    port: 9988
    app-name: my-application            # 必须与 Server 中注册的应用名一致
    server-address: {{SERVER_HOST}}:7700
    protocol: http
    store-strategy: disk                # 上下文存储：disk / memory
    max-result-length: 4096
    max-appended-wf-context-length: 4096

1
2

应用名：my-application       # 与 worker 配置的 app-name 一致
密码：{{APP_SECRET}}         # 自定义

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

@Component
public class SimpleJob implements BasicProcessor {

    @Override
    public ProcessResult process(TaskContext context) throws Exception {
        String jobParams = context.getJobParams();  // 任务参数
        long instanceId = context.getInstanceId();   // 实例 ID
        log.info("任务执行: instanceId={}, params={}", instanceId, jobParams);

        // 返回执行结果
        return new ProcessResult(true, "执行成功 " + System.currentTimeMillis());
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

@Component
public class ClearCacheJob implements BroadcastProcessor {

    @Override
    public ProcessResult broadcast(TaskContext context) {
        log.info("清理本地缓存");
        localCache.clear();
        return new ProcessResult(true);
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

@Component
public class BatchInsertJob implements MapProcessor {

    @Override
    public ProcessResult process(TaskContext context) {
        // 当前分片参数
        int subTaskId = (int) Thread.currentThread().getId();
        int totalSubTask = context.getControlPoint();   // 总分片数

        // 假设有 100 万条数据要处理
        int pageSize = 10000;
        int totalPage = 100;
        int startPage = subTaskId * (totalPage / totalSubTask);
        int endPage = (subTaskId + 1) * (totalPage / totalSubTask);

        for (int p = startPage; p < endPage; p++) {
            List<User> users = userDao.findByPage(p, pageSize);
            // 批量处理
        }

        return new ProcessResult(true, "处理 " + (endPage - startPage) + " 页");
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

@Component
public class StatisticsJob implements MapReduceProcessor {

    // 1. Map 阶段：每台 Worker 算自己分区
    @Override
    public ProcessResult process(TaskContext context) {
        List<Order> orders = orderDao.findByDateRange(...);
        long totalAmount = orders.stream().mapToLong(Order::getAmount).sum();
        return new ProcessResult(true, String.valueOf(totalAmount));
    }

    // 2. Reduce 阶段：汇总所有 Worker 结果
    @Override
    public ProcessResult reduce(TaskContext context, List<TaskResult> taskResults) {
        long total = 0;
        for (TaskResult r : taskResults) {
            total += Long.parseLong(r.getResult());
        }
        log.info("总销售额: {}", total);
        return new ProcessResult(true, "Total: " + total);
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

# 每天凌晨 2 点
time_expression: "0 0 2 * * ?"

# 每周一上午 9 点
time_expression: "0 0 9 ? * 2"

# 每 5 分钟
time_expression: "0 */5 * * * ?"

# 每月 1 号零点
time_expression: "0 0 0 1 * ?"

1
2
3

[开始] → [拉取数据] → [清洗] → [转换] → [入库] → [结束]
            ↓
         [推送通知]

1
2
3
4
5
6

订单超时关闭工作流：
1. 扫描超时订单（每 5 分钟）
2. 关闭订单
3. 释放库存
4. 退款
5. 发送通知

1
2
3
4
5

powerjob:
  worker:
    task-retry-num: 3               # 失败重试 3 次
    instance-retry-num: 0           # 实例不重试
    instance-time-limit: 0          # 0 = 不限时

1
2
3
4
5
6
7
8

# node-1
- PARAMS=--spring.profiles.active=product --oms.http.port=10010 --server.address=10.0.0.1

# node-2
- PARAMS=--spring.profiles.active=product --oms.http.port=10010 --server.address=10.0.0.2

# node-3
- PARAMS=--spring.profiles.active=product --oms.http.port=10010 --server.address=10.0.0.3

1

ERROR: app not registered

1
2
3
4

# 容器必须设置时区
environment:
  - TZ=Asia/Shanghai
  - JAVA_OPTS=-Duser.timezone=Asia/Shanghai

1
2

Server 默认单节点：1000 QPS 调度能力
3 节点集群：3000 QPS

1

固定频率 60s 任务，但任务本身耗时 80s

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

# powerjob-server.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: powerjob-server
spec:
  replicas: 3
  selector:
    matchLabels:
      app: powerjob-server
  template:
    metadata:
      labels:
        app: powerjob-server
    spec:
      containers:
      - name: server
        image: powerjob/powerjob-server:V5.0.0
        ports:
        - containerPort: 7700
        - containerPort: 10010
        env:
        - name: PARAMS
          value: "--spring.profiles.active=product
                  --spring.datasource.core.jdbc-url=jdbc:mysql://mysql:3306/powerjob
                  --spring.datasource.core.username=powerjob
                  --spring.datasource.core.password=${DB_PASSWORD}
                  --oms.http.port=10010"
        - name: TZ
          value: Asia/Shanghai
        resources:
          requests:
            memory: "2Gi"
            cpu: "1000m"
          limits:
            memory: "4Gi"
            cpu: "2000m"
---
apiVersion: v1
kind: Service
metadata:
  name: powerjob-server
spec:
  type: ClusterIP
  selector:
    app: powerjob-server
  ports:
  - name: http
    port: 7700
  - name: oms
    port: 10010

1
2
3
4
5

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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

powerjob:
  worker:
    enabled: true
    port: 9988
    app-name: my-application
    server-address: powerjob-server:7700
    protocol: http
    # V5 新增
    max-result-length: 8192
    store-strategy: disk
    # 健康检查
    health-check-interval: 30
    # 标签（V5 多租户）
    tag: production
    namespace: tenant-a

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

# 订单超时关闭工作流（V5 新增循环节点）
nodes:
  - id: scan_orders
    type: Basic
    processor: scanTimeoutOrders

  - id: close_order
    type: Basic
    processor: closeOrder
    dependsOn: [scan_orders]

  - id: release_stock
    type: Basic
    processor: releaseStock
    dependsOn: [close_order]

  - id: refund
    type: Basic
    processor: refundOrder
    dependsOn: [release_stock]

  - id: notify_user
    type: Basic
    processor: sendNotification
    dependsOn: [refund]
    # V5 新增：失败重试 + 告警
    retryCount: 3
    alertEnabled: true
    alertChannels: [dingtalk, feishu]

1
2
3
4
5
6
7
8
9

# application.yml（Server）
management:
  endpoints:
    web:
      exposure:
        include: health,info,prometheus
  metrics:
    tags:
      application: powerjob-server

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

中型互联网业务（Java + Python 异构）
├── PowerJob Server V5 × 3 副本（K8s Deployment）
├── MySQL 8.4 LTS 存储（StatefulSet + 持久卷）
├── PowerJob Worker × N 副本（与应用一起跑）
│   ├── Java 微服务 Worker
│   ├── Python 数据同步 Worker（HTTP 类型）
│   └── Shell 清理任务 Worker
├── DAG 工作流编排业务（订单 / 报表 / 同步）
├── Prometheus + Grafana 监控
└── 告警：钉钉 + 飞书 + 企业微信