基于Zookeeper的任务调度系统的设计方案和代码实现网站首页 技术杂谈

基于Zookeeper的任务调度系统的设计方案和代码实现

星辰@Sea 2025-02-11 12:01:03

简介基于Zookeeper的任务调度系统的设计方案和代码实现

引言

随着微服务架构的普及，任务调度变得越来越重要。现有系统中，任务调度的可靠性、扩展性和效率需要进一步提升。基于Zookeeper的任务调度系统能够提供高可用性和强一致性，适用于复杂的任务管理需求。

项目目标
任务动态注册与发现：支持任务的动态注册与发现，确保任务能够灵活地加入和退出系统。
负载均衡与容错机制：实现负载均衡策略，确保任务执行的公平性；同时支持任务失败后的重试和补偿机制。
灵活性与可扩展性：允许用户根据需求调整任务参数，支持多种任务类型和执行环境。
高可用性：基于Zookeeper的分布式协调机制，确保任务注册中心的高可用性和强一致性。

系统架构设计

1. 任务管理服务

职责：负责任务的注册、管理、发现、更新和状态管理。
接口：TaskManagerInterface
public interface TaskManagerInterface {
void registerTask(Task task);
void unregisterTask(String taskId);
List discoverTasks();
void updateTaskStatus(String taskId, TaskStatus status);
}

2. 任务调度器

职责：根据任务状态和负载，动态分配任务；支持负载均衡策略（轮询或加权）。
实现类：ZookeeperTaskManager
public class ZookeeperTaskManager implements TaskManagerInterface {
private CuratorFramework curator;
private String basePath;

public ZookeeperTaskManager(CuratorFramework curator, String basePath) {
    this.curator = curator;
    this.basePath = basePath;
}

@Override
public void registerTask(Task task) throws Exception {
    String taskPath = basePath + "/task_" + task.getId();
    curator.create().forPath(taskPath, serializeTask(task));
}

@Override
public void unregisterTask(String taskId) throws Exception {
    String taskPath = basePath + "/task_" + taskId;
    curator.delete().forPath(taskPath);
}

@Override
public List<Task> discoverTasks() throws Exception {
    List<Task> tasks = new ArrayList<>();
    List<String> taskPaths = curator.getChildren().forPath(basePath);
    for (String taskPath : taskPaths) {
        try {
            Task task = deserializeTask(curator.getData().forPath(taskPath));
            tasks.add(task);
        } catch (Exception e) {
            System.err.println("Error deserializing task: " + e.getMessage());
        }
    }
    return tasks;
}

@Override
public void updateTaskStatus(String taskId, TaskStatus status) throws Exception {
    String taskPath = basePath + "/task_" + taskId;
    try {
        byte[] data = serializeTask(task);
        curator.create().forPath(taskPath).write(data);
    } catch (Exception e) {
        System.err.println("Error updating task status: " + e.getMessage());
    }
}

}

3. 任务执行器

职责：具体执行任务，支持任务重试和补偿机制。
实现类：TaskExecutor
public class TaskExecutor {
private ExecutorService executor;
private Runnable taskRunner;
private Throwable exception;
private int retriesLeft;

public TaskExecutor(ExecutorService executor, Runnable taskRunner) {
    this.executor = executor;
    this.taskRunner = taskRunner;
}

public void executeTask(Task task) throws Exception {
    retriesLeft = task.getRetries() - 1;
    try {
        if (retriesLeft <= 0) {
            throw new IllegalStateException("Task has already been retried " + task.getRetries() + " times");
        }
        executor.submit(taskRunner.run(task));
    } catch (Exception e) {
        if (retriesLeft > 0) {
            task.setRetries(task.getRetries() + 1);
            task.setException(e);
            taskManager.updateTaskStatus(task.getId(), TaskStatus.RETRYING);
            executor.submit(taskRunner.run(task));
        } else {
            task.setRetries(0);
            task.setException(e);
            taskManager.updateTaskStatus(task.getId(), TaskStatus.FAILED);
        }
    }
}

}

4. 数据库设计

任务表 (task)

task_id (主键)：任务唯一标识符。
task_type：任务类型。
execution_params：执行参数。
state：任务状态。
execution_time：任务的执行时间。
retries：任务重试次数。
weight：任务优先级。
updated_time：任务最后更新时间。
CREATE TABLE task (
task_id VARCHAR(255) NOT NULL AUTO_INCREMENT,
task_type VARCHAR(255) NOT NULL,
execution_params JSONB NOT NULL,
state VARCHAR(255) DEFAULT ‘PENDING’,
execution_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
retries INT DEFAULT 0,
weight INT DEFAULT 1,
updated_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (task_id),
UNIQUE KEY execution_id (task_id, created_at)
);

任务执行记录表 (task_execution)

execution_id (主键)：任务执行记录的唯一标识符。
task_id：关联的任务ID。
status：任务执行状态。
error：任务执行错误信息。
weight：任务优先级。
execution_time：任务执行时间。
updated_time：任务执行记录的更新时间。
CREATE TABLE task_execution (
execution_id VARCHAR(255) NOT NULL AUTO_INCREMENT,
task_id VARCHAR(255) NOT NULL,
status VARCHAR(255) DEFAULT ‘PENDING’,
error TEXT,
weight INT DEFAULT 1,
execution_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (execution_id),
UNIQUE KEY execution_id
);

任务依赖表 (task_dependency)

execution_id (主键)：任务执行记录的唯一标识符。
task_id：依赖的任务ID。
type：任务依赖类型（如 SUCCESS, FAILED）。
updated_time：任务依赖关系的更新时间。
CREATE TABLE task_dependency (
execution_id VARCHAR(255) NOT NULL AUTO_INCREMENT,
task_id VARCHAR(255) NOT NULL,
type VARCHAR(255) DEFAULT ‘NONE’,
updated_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (execution_id),
UNIQUE KEY execution_id
);

任务依赖执行表 (task_execution_dependency)

execution_id (主键)：任务执行记录的唯一标识符。
dependency_id：依赖任务的执行记录的唯一标识符。
type：任务依赖类型（如 Jeremy, failed）。
updated_time：任务依赖关系的更新时间。
CREATE TABLE task_execution_dependency (
execution_id VARCHAR(255) NOT NULL AUTO_INCREMENT,
dependency_id VARCHAR(255) NOT NULL,
type VARCHAR(255) DEFAULT ‘NONE’,
updated_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (execution_id),
UNIQUE KEY execution_id
);

服务注册表 (service)

service_id (主键)：服务唯一标识符。
ip_address：服务的IP地址。
port：服务的端口。
status：服务状态（UP, DOWN）。
created_at：服务的创建时间。
updated_at：服务的最后更新时间。
CREATE TABLE service (
service_id VARCHAR(255) NOT NULL AUTO_INCREMENT,
ip_address VARCHAR(255) NOT NULL,
port INT DEFAULT 8080,
status VARCHAR(255) DEFAULT ‘UP’,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (service_id),
UNIQUE KEY service_id
);

系统架构图

graph TD
+/----+ CuratorFramework [基于Zookeeper的任务调度系统]
+/----+ TaskManager
| -->| discoverTasks
| -->| registerTask
| -->| updateTaskStatus
+/----+ TaskScheduler
| -->| taskAllocation
| -->| loadBalancing
+/----+ TaskExecutor
| -->| taskExecution
| -->| taskDependency
| -->| taskExecutionDependency
+/----+ ServiceRegistry
| -->| serviceRegistration
| -->| serviceDiscovery