作业监听器

可通过配置多个任务监听器，在任务执行前和执行后执行监听的方法。监听器分为每台作业节点均执行和分布式场景中仅单一节点执行2种。

1. 每台作业节点均执行的监听

若作业处理作业服务器的文件，处理完成后删除文件，可考虑使用每个节点均执行清理任务。此类型任务实现简单，且无需考虑全局分布式任务是否完成，请尽量使用此类型监听器。

步骤：

  • 定义监听器
  2. public class MyElasticJobListener implements ElasticJobListener {
  4.     @Override
  5.     public void beforeJobExecuted(ShardingContexts shardingContexts) {
  6.         // do something ...
  7.     }
  9.     @Override
  10.     public void afterJobExecuted(ShardingContexts shardingContexts) {
  11.         // do something ...
  12.     }
  13. }
  • 将监听器作为参数传入JobScheduler
  1. public class JobMain {
  3.     public static void main(String[] args) {
  4.         new JobScheduler(createRegistryCenter(), createJobConfiguration(), new MyElasticJobListener()).init();
  5.     }
  7.     private static CoordinatorRegistryCenter createRegistryCenter() {
  8.         CoordinatorRegistryCenter regCenter = new ZookeeperRegistryCenter(new ZookeeperConfiguration("zk_host:2181", "elastic-job-demo"));
  9.         regCenter.init();
  10.         return regCenter;
  11.     }
  13.     private static LiteJobConfiguration createJobConfiguration() {
  14.         // 创建作业配置
  15.         ...
  16.     }
  17. }

2. 分布式场景中仅单一节点执行的监听

若作业处理数据库数据，处理完成后只需一个节点完成数据清理任务即可。此类型任务处理复杂，需同步分布式环境下作业的状态同步，提供了超时设置来避免作业不同步导致的死锁，请谨慎使用。

步骤：

  • 定义监听器
  2. public class TestDistributeOnceElasticJobListener extends AbstractDistributeOnceElasticJobListener {
  4.     public TestDistributeOnceElasticJobListener(long startTimeoutMills, long completeTimeoutMills) {
  5.         super(startTimeoutMills, completeTimeoutMills);
  6.     }
  8.     @Override
  9.     public void doBeforeJobExecutedAtLastStarted(ShardingContexts shardingContexts) {
  10.         // do something ...
  11.     }
  13.     @Override
  14.     public void doAfterJobExecutedAtLastCompleted(ShardingContexts shardingContexts) {
  15.         // do something ...
  16.     }
  17. }
  • 将监听器作为参数传入JobScheduler
  1. public class JobMain {
  3.     public static void main(String[] args) {
  4.         long startTimeoutMills = 5000L;
  5.         long completeTimeoutMills = 10000L;
  6.         new JobScheduler(createRegistryCenter(), createJobConfiguration(), new MyDistributeOnceElasticJobListener(startTimeoutMills, completeTimeoutMills)).init();
  7.     }
  9.     private static CoordinatorRegistryCenter createRegistryCenter() {
  10.         CoordinatorRegistryCenter regCenter = new ZookeeperRegistryCenter(new ZookeeperConfiguration("zk_host:2181", "elastic-job-demo"));
  11.         regCenter.init();
  12.         return regCenter;
  13.     }
  15.     private static LiteJobConfiguration createJobConfiguration() {
  16.         // 创建作业配置
  17.         ...
  18.     }
  19. }