分布式定时任务 elatic-job 自定义注解使用

痛定思痛。 2021-09-18 04:18 253阅读 0赞

以前使用 elastic-job 都是使用 \*.xml 配置文件的形式，但是现在的 springboot 项目都倾向于使用 Java Bean 的形式，所以本文介绍了如何使用Java Bean 的形式使用 elastic-job

1.  首先是 elastic-job maven 依赖  
    关于 springboot 的依赖这里省略掉了。

<dependency>
    <groupId>com.dangdang</groupId>
    	<artifactId>elastic-job-lite-core</artifactId>
    	<version>2.1.5</version>
    </dependency>
    <dependency>
    	<groupId>com.dangdang</groupId>
    	<artifactId>elastic-job-lite-spring</artifactId>
    	<version>2.1.5</version>
    </dependency>

1.  ElasticJobConfig 配置类  
    zookeeper 的配置信息放在 application.properties  
    该配置类主要配置了 elastic-job 的注册到 zookeeper 需要的一些bean

@Configuration
    public class ElasticJobConfig {
    
        @Value("${elasticjob.zookeeper.list}")
        private String serverlists;
        @Value("${elasticjob.zookeeper.namespace}")
        private String namespace;
    
        /**
         * zookeeper config
         * 
         * @return
         */
        @Bean
        public ZookeeperConfiguration zkConfig() {
            ZookeeperConfiguration config = new ZookeeperConfiguration(serverlists, namespace);
            config.setBaseSleepTimeMilliseconds(1000);
            config.setMaxRetries(3);
            config.setMaxSleepTimeMilliseconds(3000);
            config.setConnectionTimeoutMilliseconds(3000);
            return config;
        }
    
        /**
         * 初始化注册
         * 
         * @param config
         * @return
         */
        @Bean(initMethod = "init")
        public ZookeeperRegistryCenter regCenter(ZookeeperConfiguration config) {
            return new ZookeeperRegistryCenter(config);
        }
    }

application.properties

elasticjob.zookeeper.list=localhost:2181
    elasticjob.zookeeper.namespace=test

1.  使用自定义注解来配置定时任务

@Target(ElementType.TYPE)
    @Retention(RetentionPolicy.RUNTIME)
    @Documented
    @Inherited
    @Component
    public @interface Task {
    
        /**
         * description: 任务名称
         * 
         * @return
         */
        @NotNull
        String jobName();
    
        /**
         * description: Cron 表达式
         * 
         * @return
         */
        @NotNull
        String cron();
    
        /**
         * description: 描述
         * 
         * @return
         */
        @NotNull
        String desc();
    
        /**
         * description: 是否快速失败
         * 
         * @return
         */
        boolean failover() default false;
    
        /**
         * description: sharing 参数
         * 
         * @return
         */
        String shardingItemParameters() default "";
    
        /**
         * description: 任务参数
         * 
         * @return
         */
        String jobParameter() default "";
    
        boolean overwrite() default true;
    
        /**
         * description: 几个分片执行（默认是 1）
         * 
         * @return
         */
        int shardingTotalCount() default 1;
    
    }

1.  注册定时任务  
    通过实现 Spring 的 ApplicationContextAware 来获取 ApplicationContext 上下文信息，获取注册到Spring中的Bean  
    通过实现 InitializingBean 在给Bean注册到 Spring 之后，利用 ApplicationContext 获取使用 @Task 注解的 Bean，并注册到 elastic-job 中

@Component
    @Slf4j
    public class RegisterJobCenter implements ApplicationContextAware, InitializingBean {
    
        private ApplicationContext applicationContext;
    
        @Autowired
        private ZookeeperRegistryCenter regCenter;
    
        @Override
        public void afterPropertiesSet() throws Exception {
            // 获取 注解 Task 的Bean
            Map<String, Object> registerJobs = applicationContext.getBeansWithAnnotation(Task.class);
            // 开始遍历
            for (Map.Entry<String, Object> entry : registerJobs.entrySet()) {
                try {
                    Object object = entry.getValue();
                    if (!(object instanceof ElasticJob)) {
                        throw new ClassCastException("[" + object.getClass().getName()
                                + "] The class type is not com.dangdang.ddframe.job.api.ElasticJob");
                    }
                    Task task = AnnotationUtils.findAnnotation(object.getClass(), Task.class);
                    SpringJobScheduler springJobScheduler = new SpringJobScheduler((ElasticJob) object, regCenter,
                            getJobConfiguration(task, object));
                    springJobScheduler.init();
                    log.info("JobName: {} Register Successfully .", task.jobName());
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                }
            }
        }
    
        /**
         * description: 构建 Job 配置
         * 
         * @param taskJob
         * @param object
         * @return
         */
        private LiteJobConfiguration getJobConfiguration(Task taskJob, Object object) {
            // 参数校验
            Optional.ofNullable(taskJob.jobName())
                    .orElseThrow(() -> new IllegalArgumentException("The jobName cannot be null !"));
            Optional.ofNullable(taskJob.cron())
                    .orElseThrow(() -> new IllegalArgumentException("The cron cannot be null !"));
            Optional.ofNullable(taskJob.desc())
                    .orElseThrow(() -> new IllegalArgumentException("The desc cannot be null !"));
            // 构建配置
            SimpleJobConfiguration simpleJobConfiguration = new SimpleJobConfiguration(JobCoreConfiguration
                    .newBuilder(taskJob.jobName(), taskJob.cron(), taskJob.shardingTotalCount())
                    .shardingItemParameters(
                            StringUtils.isEmpty(taskJob.shardingItemParameters()) ? null : taskJob.shardingItemParameters())
                    .description(taskJob.desc()).failover(taskJob.failover())
                    .jobParameter(StringUtils.isEmpty(taskJob.jobParameter()) ? null : taskJob.jobParameter()).build(),
                    object.getClass().getName());
            return LiteJobConfiguration.newBuilder(simpleJobConfiguration).overwrite(taskJob.overwrite())
                    .monitorExecution(true).build();
        }
    
        @Override
        public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
            this.applicationContext = applicationContext;
        }
    
    }

到此，基于 Spring Bean 的基本结构就完成了，接下来就是使用其实现我们的定时任务了。

Example:

@Task(jobName = "jobA", cron = "0/5 * * * * ? ", desc = "JobA")
    public class JobA implements SimpleJob {
    
        @Override
        public void execute(ShardingContext shardingContext) {
            System.out.println("JobA is running ...")
        }
    }
    ------------------------------------
    JobA is running ...
    JobA is running ...
    JobA is running ...