Agent Java项目入门指南：从零搭建高可用的分布式任务调度系统

1次阅读

共计 2871 个字符，预计需要花费 8 分钟才能阅读完成。

在传统单机定时任务（如 Spring 自带的 @Scheduled）中，我们经常会遇到两个致命问题：

单点故障 ：如果部署定时任务的服务器宕机，所有任务都会停止运行
任务重复执行 ：在集群环境下，如果不加控制，同一个任务可能被多个节点同时触发

我曾在一个电商项目中遇到过惨痛教训：促销活动的库存同步任务因为单节点崩溃而停滞，导致前端页面显示库存与实际相差 30%。这促使我开始研究分布式任务调度方案。

特性	Quartz	XXL-JOB	Elastic-Job	Agent 架构
分布式支持	弱 (需 DB 锁)	强	强	强
可视化界面	无	有	有	可扩展
任务分片	不支持	支持	支持	支持
故障转移	手动实现	自动	自动	自动 + 智能

选择 Agent 架构的关键原因：

与 Spring Boot 生态无缝集成
基于 ZooKeeper 的协调机制比数据库轮询更高效
灵活的扩展点设计，方便二次开发

在 Spring Boot 项目中只需两步：

添加 starter 依赖（注意版本号）：

<dependency>
    <groupId>com.agent</groupId>
    <artifactId>agent-spring-boot-starter</artifactId>
    <version>2.1.4</version>
</dependency>

在 application.yml 中配置 ZooKeeper 地址：

agent:
  zookeeper:
    servers: 192.168.1.100:2181,192.168.1.101:2181
    namespace: /agent-cluster

@Slf4j
public class JobRegistry {
    // 使用 CuratorFramework 与 ZK 交互
    @Autowired
    private CuratorFramework client;

    /**
     * 注册任务到 ZK 持久节点
     * @throws Exception 当 ZK 连接异常时抛出
     * @see org.apache.curator.framework.api.CuratorEvent
     */
    public void register(JobDefinition job) throws Exception {String path = "/jobs/" + job.getJobId();
        client.create().creatingParentsIfNeeded()
              .withMode(CreateMode.PERSISTENT)
              .forPath(path, job.toString().getBytes());
    }
}

关键点在于使用 ZK 的 Watcher 机制实现事件监听：

// 监听任务节点变化
public void watchJobNodes() {PathChildrenCache cache = new PathChildrenCache(client, "/jobs", true);
    cache.getListenable().addListener((client, event) -> {switch (event.getType()) {
            case CHILD_ADDED:
                scheduleJob(event.getData());
                break;
            case CHILD_REMOVED:
                cancelJob(event.getData());
                break;
        }
    });
}

建议采用线程池隔离不同优先级的任务：

@Bean
public Executor jobExecutor() {ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(10);
    executor.setMaxPoolSize(50);
    executor.setQueueCapacity(100);
    executor.setThreadNamePrefix("job-exec-");
    executor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy());
    return executor;
}

故障转移的核心逻辑：

// 在 ZK 上创建临时节点表示活跃节点
String leaderPath = "/leader";
client.create().withMode(CreateMode.EPHEMERAL)
      .forPath(leaderPath, instanceId.getBytes());

// 设置监听
client.getData().usingWatcher((Watcher) event -> {if (event.getType() == EventType.NodeDeleted) {electNewLeader(); // 重新选举
    }
}).forPath(leaderPath);

方案	查询性能	存储成本	实时性	适合场景
ELK	高	高	秒级	需要复杂日志分析
Grafana	中	中	分钟级	监控告警为主

建议：关键业务日志用 ELK，性能指标用 Grafana

为每个任务组设置独立的线程池
使用 Kubernetes 的 HPA 自动扩缩容
通过 ZK 的临时节点感知节点变化

@Bean
public MBeanServer mBeanServer() {MBeanServerFactoryBean factory = new MBeanServerFactoryBean();
    factory.setLocateExistingServerIfPossible(true);
    return factory.getObject();}

@Bean
public Exporter exporter() {AnnotationMBeanExporter exporter = new AnnotationMBeanExporter();
    exporter.setServer(mBeanServer());
    return exporter;
}

当任务执行时间超过调度间隔时，我的解决方案是：