SpringBoot2整合ElasticJob框架过程详解

<dependency>
  <groupId>com.dangdang</groupId>
  <artifactId>elastic-job-lite-core</www.lanboylsy.com artifactId>
  <version>2.1.5</version>
</dependency>
<dependency>
  <groupId>com.dangdang<www.yuanyangyul.com /groupId>
  <artifactId>elastic-job-lite-spring<www.lexuancaizc.cn /artifactId>
  <version>2.1.5</version>
</dependency>

zookeeper:
 server: 127.0.0.1:2181
 namespace: es-job
job-config:
 cron: 0/10 * * * * ?
 shardCount: 1
 shardItem: 0=A,1=B,2=shentuylzc.cn C,3www.yongxinylzn.cn=D

@Inherited
@Target({ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
public @interface TaskJobSign www.zhuyngyule.cn{

  @AliasFor("cron"www.feiyuptzc.cn)
  String value(www.yinmao2zhuce.cn) default "";

  @AliasFor("value")
  String cron(www.wujiu5zhuce.cn) default "";

  String jobName(www.shengyunyule.cn) default "";

}

@Component
@TaskJobSign(cron = www.anxing4zc.cn"0/5 * * * * ?",jobName =www.jucaiyle.cn "Hello-Job")
public class HelloJob implements SimpleJob {

  private static final Logger LOG = LoggerFactory.getLogger(HelloJob.class.getName()) ;

  @Override
  public void execute(ShardingContext shardingContext) {
    LOG.info("当前线程: "+Thread.currentThread().getId());
    LOG.info("任务分片："+shardingContext.getShardingTotalCount());
    LOG.info("当前分片："+shardingContext.getShardingItem());
    LOG.info("分片参数："+shardingContext.getShardingParameter());
    LOG.info("任务参数："+shardingContext.getJobParameter());
  }
}

@Configuration
public class ElasticJobConfig {

  @Resource
  private ApplicationContext applicationContext ;
  @Resource
  private ZookeeperRegistryCenter zookeeperRegistryCenter;

  @Value("${job-config.cron}") private String cron ;
  @Value("${job-config.shardCount}"www.jucaiylzc.cn) private int shardCount ;
  @Value("${job-config.shardItem}") private String shardItem ;
  
  /**
   * 配置任务监听器
   */
  @Bean
  public ElasticJobListener elasticJobListener() {
    return new TaskJobListener();
  }
  /**
   * 初始化配置任务
   */
  @PostConstruct
  public void initTaskJob() {
    Map<String, SimpleJob> jobMap = this.applicationContext.getBeansOfType(SimpleJob.class);
    Iterator iterator = jobMap.entrySet().iterator();
    while (iterator.hasNext()) {
      // 自定义注解管理
      Map.Entry<String, SimpleJob> entry = (Map.Entry)iterator.next();
      SimpleJob simpleJob = entry.getValue();
      TaskJobSign taskJobSign = simpleJob.getClass().getAnnotation(TaskJobSign.class);
      if (taskJobSign != null){
        String cron = taskJobSign.cron() ;
        String jobName = taskJobSign.jobName() ;
        // 生成配置
        SimpleJobConfiguration simpleJobConfiguration = new SimpleJobConfiguration(
                        JobCoreConfiguration.newBuilder(jobName, cron, shardCount)
                        .shardingItemParameters(shardItem).jobParameter(jobName).build(),
                        simpleJob.getClass().getCanonicalName());
        LiteJobConfiguration liteJobConfiguration = LiteJobConfiguration.newBuilder(
                        simpleJobConfiguration).overwrite(true).build();
        TaskJobListener taskJobListener = new TaskJobListener();
        // 初始化任务
        SpringJobScheduler jobScheduler = new SpringJobScheduler(
                        simpleJob, zookeeperRegistryCenter,
                        liteJobConfiguration, taskJobListener);
        jobScheduler.init();
      }
    }
  }
}

@Component
public class TaskJobListener implements ElasticJobListener {
  private static final Logger LOG = LoggerFactory.getLogger(TaskJobListener.class);

  private long beginTime = 0;

  @Override
  public void beforeJobExecuted(ShardingContexts shardingContexts) {
    beginTime = System.currentTimeMillis();
    LOG.info(shardingContexts.getJobName()+"===>开始...");
  }

  @Override
  public void afterJobExecuted(ShardingContexts shardingContexts) {
    long endTime = System.currentTimeMillis();
    LOG.info(shardingContexts.getJobName()+
    "===>结束...[耗时："+(endTime - beginTime)+"]");
  }
}

@Component
public class GetTimeJob implements SimpleJob {

  private static final Logger LOG = LoggerFactory.getLogger(GetTimeJob.class.getName()) ;

  private static final SimpleDateFormat format =
      new SimpleDateFormat("yyyy-MM-dd HH:mm:ss") ;

  @Override
  public void execute(ShardingContext shardingContext) {
    LOG.info("Job Name:"+shardingContext.getJobName());
    LOG.info("Local Time:"+format.format(new Date()));
  }
}

@Service
public class TaskJobService {

  @Resource
  private ZookeeperRegistryCenter zookeeperRegistryCenter;

  public void addTaskJob(final String jobName,final SimpleJob simpleJob,
              final String cron,final int shardCount,final String shardItem) {
    // 配置过程
    JobCoreConfiguration jobCoreConfiguration = JobCoreConfiguration.newBuilder(
                          jobName, cron, shardCount)
                          .shardingItemParameters(shardItem).build();
    JobTypeConfiguration jobTypeConfiguration = new SimpleJobConfiguration(jobCoreConfiguration,
                          simpleJob.getClass().getCanonicalName());
    LiteJobConfiguration liteJobConfiguration = LiteJobConfiguration.newBuilder(
                          jobTypeConfiguration).overwrite(true).build();
    TaskJobListener taskJobListener = new TaskJobListener();
    // 加载执行
    SpringJobScheduler jobScheduler = new SpringJobScheduler(
        simpleJob, zookeeperRegistryCenter,
        liteJobConfiguration, taskJobListener);
    jobScheduler.init();
  }

}

@RestController
public class TaskJobController {

  @Resource
  private TaskJobService taskJobService ;

  @RequestMapping("/addJob")
  public String addJob(@RequestParam("cron") String cron,@RequestParam("jobName") String jobName,
             @RequestParam("shardCount") Integer shardCount,
             @RequestParam("shardItem") String shardItem) {
    taskJobService.addTaskJob(jobName, new GetTimeJob(), cron, shardCount, shardItem);
    return "success";
  }
}