four-level-schedule

四级调度系统 (Four-Level Schedule)

项目简介

四级调度系统是一个基于Spring Boot和Kafka的分布式任务调度系统，通过Producer-Consumer模式实现任务的生产与消费，并配备Manager模块进行实例管理和动态扩缩容。系统支持Docker容器化部署，能够根据任务积压情况和资源负载自动调整实例数量，实现高效的资源利用和任务处理。

系统架构

系统由四个核心模块组成：

1. schedule-producer: 任务生产者模块，负责创建和发送任务到Kafka消息队列
2. schedule-consumer: 任务消费者模块，负责从Kafka消费并处理任务
3. schedule-manager: 管理模块，负责监控实例状态、管理Docker容器和动态扩缩容
4. schedule-monitor: Kafka监控模块，提供Topic和消费组的监控管理功能

技术栈

• Java 8
• Spring Boot 2.7.18
• Spring Kafka 2.8.11
• Docker (通过Docker Java API)
• Fastjson 1.2.83
• Lombok 1.18.30

模块说明

schedule-producer

任务生产者模块，提供RESTful API用于创建和发送任务。

主要功能：

• 接收任务请求
• 将任务序列化并发送到Kafka主题
• 支持任务重试机制

配置说明：

• 服务端口: 8081
• Kafka服务器: 10.192.72.13:9092
• 默认主题: test-topic (6个分区)

schedule-consumer

任务消费者模块，监听Kafka主题并消费任务。

主要功能：

• 监听Kafka主题
• 消费并处理任务
• 支持手动提交offset

配置说明：

• 服务端口: 8082
• Kafka消费者组: yusp-topic-group
• 监听主题: test-topic

schedule-manager

管理模块，负责监控和管理解析服务实例。

主要功能：

• 监控实例健康状态
• 管理Docker容器生命周期
• 根据任务积压和资源负载动态扩缩容
• GPU负载监控

配置说明：

• 服务端口: 8083
• Docker守护进程地址: tcp://127.0.0.1:2375
• 心跳超时阈值: 30000ms
• 最小活跃实例数: 3
• 最大活跃实例数: 10
• 任务积压阈值: 100
• GPU负载阈值: 80%

schedule-monitor

Kafka监控模块，提供Topic和消费组的监控管理功能。

主要功能：

• Topic管理：获取Topic列表（支持分页）、创建Topic、删除Topic、修改Topic分区
• Topic详情：查询指定Topic的详细信息，包括分区、消息数、副本数等
• 分区消息查询：获取指定分区的消息列表
• 消费组管理：获取消费组列表、消费组详情、删除消费组
• 消费组监控：显示消费组成员、消费偏移量、消费延迟等信息

配置说明：

• 服务端口: 8094
• Kafka服务器: 10.192.72.13:9092
• 监控消费者组: monitor-group

环境要求

• JDK 8+
• Maven 3.6+
• Docker (用于Manager模块)
• Kafka 2.8+

快速开始

1. 克隆项目

git clone <repository-url>
cd four-level-schedule

2. 配置Kafka连接

修改各模块的application.yml文件，配置Kafka服务器地址：

spring:
  kafka:
    bootstrap-servers: <your-kafka-server>:9092

3. 配置Docker连接

修改schedule-manager/src/main/resources/application.yml，配置Docker守护进程地址：

docker:
  host: tcp://<your-docker-host>:2375

4. 编译项目

mvn clean package

5. 启动服务

分别启动四个模块：

# 启动Producer
java -jar schedule-producer/target/schedule-producer.jar
nohup java -jar schedule-producer.jar > producer.log 2>&1 &


# 启动Consumer
java -jar schedule-consumer/target/schedule-consumer.jar
nohup java -jar schedule-consumer-1.0-SNAPSHOT.jar \
  --spring.config.location=./application.yml > consumer.log 2>&1 &

# 启动Manager
java -jar schedule-manager/target/schedule-manager.jar
java -jar schedule-manager-1.0-SNAPSHOT.jar \
  --spring.config.location=./application.yml   > manager.log 2>&1 &
# 启动Monitor
java -jar schedule-monitor/target/schedule-monitor.jar

解析服务部署：

启动manager
在 /home/app/four-level-schedule/jar/schedule-manager/schedule-manager-1.0-SNAPSHOT.jar 目录下执行
java -jar schedule-manager-1.0-SNAPSHOT.jar \
  --spring.config.location=./application.yml   > manager.log 2>&1 &
  
  
 启动Consumer
 在/home/app/four-level-schedule/jar/schedule-consumer/schedule-consumer-1.0-SNAPSHOT.jar 执行
 nohup java -jar schedule-consumer-1.0-SNAPSHOT.jar \
  --spring.config.location=./application.yml > consumer.log 2>&1 &

测试指南

发送任务

通过Producer模块提供的RESTful API发送任务：

POST http://localhost:8081/api/send

测试数据
{
    "topic":"schedule-topic",
    "key":"1",
    "message":"./examples/test1.pdf"
}

监控任务

Manager模块会自动监控任务处理情况和实例状态，并根据配置的阈值进行扩缩容。

Kafka监控

通过Monitor模块提供的RESTful API监控Kafka：

• 获取Topic列表（分页）:

  curl "http://localhost:8094/api/kafka/monitor/topics?page=1&pageSize=10"
  

• 获取Topic详情:

  curl "http://localhost:8094/api/kafka/monitor/topics/test-topic"
  

• 创建Topic:

  curl -X POST "http://localhost:8094/api/kafka/monitor/topics" -H "Content-Type: application/json" -d "{\"topicName\":\"new-topic\",\"partitions\":3,\"replicationFactor\":1}"
  

• 删除Topic:

  curl -X DELETE "http://localhost:8094/api/kafka/monitor/topics/test-topic"
  

• 修改Topic分区:

  curl -X POST "http://localhost:8094/api/kafka/monitor/topics/test-topic/partitions" -H "Content-Type: application/json" -d "{\"partitions\":5}"
  

• 获取分区消息:

  curl "http://localhost:8094/api/kafka/monitor/topics/test-topic/partitions/0/messages?startOffset=0&maxRecords=10"
  

• 获取消费组列表:

  curl "http://localhost:8094/api/kafka/monitor/consumer-groups"
  

• 获取消费组详情:

  curl "http://localhost:8094/api/kafka/monitor/consumer-groups/group-1"
  

• 删除消费组:

  curl -X DELETE "http://localhost:8094/api/kafka/monitor/consumer-groups/group-1"
  

配置说明

Kafka配置

所有模块都使用相同的Kafka配置，包括bootstrap-servers、序列化器等。可根据实际环境调整。

Manager监控配置

Manager模块提供了丰富的监控配置选项：

• heartbeat-timeout: 心跳超时阈值
• min-active-instance: 最小活跃实例数
• max-active-instance: 最大活跃实例数
• task-backlog-threshold: 任务积压阈值
• gpu-load-threshold: GPU负载阈值

监控模块配置 (schedule-monitor/src/main/resources/application.yml)

server:
  port: 8094

spring:
  application:
    name: schedule-monitor
  kafka:
    bootstrap-servers: 10.192.72.13:9092
    consumer:
      group-id: monitor-group
      enable-auto-commit: true
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

项目结构

four-level-schedule/
├── pom.xml                          # 父POM文件
├── schedule-producer/               # 任务生产者模块
│   ├── pom.xml
│   └── src/
│       ├── main/
│       │   ├── java/
│       │   └── resources/
│       └── test/
├── schedule-consumer/               # 任务消费者模块
│   ├── pom.xml
│   └── src/
│       ├── main/
│       │   ├── java/
│       │   └── resources/
│       └── test/
├── schedule-manager/                # 管理模块
│   ├── pom.xml
│   ├── doc/                         # 文档目录
│   └── src/
│       ├── main/
│       │   ├── java/
│       │   └── resources/
│       └── test/
└── schedule-monitor/                # Kafka监控模块
    ├── pom.xml
    └── src/
        ├── main/
        │   ├── java/
        │   └── resources/
        └── test/

常见问题

1. Kafka连接失败

检查Kafka服务器地址和端口是否正确，确保Kafka服务正常运行。

2. Docker连接失败

检查Docker守护进程是否启用，确保Docker API端口(2375)已开放。

3. 实例无法自动扩缩容

检查Manager模块的监控配置是否合理，确保任务积压和资源负载达到阈值。

许可证

本项目采用内部许可证，未经授权不得用于商业用途。

联系方式

如有问题或建议，请联系项目维护者。