SpringBoot 集成 elastic job 3

官方的starter目前不支持注解声明job。
这里做了一些扩展

  • 支持注解声明job
  • 支持多级注解,一个service声明多个job
  • 支持手工增加,删除,触发job

https://github.com/kangarooxin/elastic-job3-spring-boot-starter

使用说明:

  1. 在pom.xml 添加引用
<dependency>
    <groupId>io.github.kangarooxin</groupId>
    <artifactId>elastic-job3-spring-boot-starter</artifactId>
    <version>1.0.1</version>
</dependency>
  1. 配置
   #reg-center
   elasticjob.reg-center.server-lists=localhost:2181
   elasticjob.reg-center.namespace=${spring.application.name}.elastic-job
  1. 使用注解
@Service
@ElasticJobScheduler(cron = "0/5 * * * * ?", shardingTotalCount = 4, name = "SimpleElasticJob",
        shardingItemParameters = "0=0,1=0,2=1,3=1", jobParameters = "parameter")
@Slf4j
public class SimpleElasticJob implements SimpleJob {

   @Override
   public void execute(ShardingContext shardingContext) {
      log.info("Thread ID: {}, ShardingTotalCount: {}, ShardingItem: {}, ShardingParameter: {}, JobName: {}, JobParameter: {}",
              Thread.currentThread().getId(),
              shardingContext.getShardingTotalCount(),
              shardingContext.getShardingItem(),
              shardingContext.getShardingParameter(),
              shardingContext.getJobName(),
              shardingContext.getJobParameter()
      );
   }
}
  1. 支持多级注解
@Service
@ElasticJobScheduler(cron = "0/5 * * * * ?", shardingTotalCount = 4, name = "SimpleElasticJobLevel1",
        shardingItemParameters = "0=0,1=0,2=1,3=1", jobParameters = "parameter")
@ElasticJobScheduler(cron = "0 */1 * * * ?", shardingTotalCount = 4, name = "SimpleElasticJobLevel2",
        shardingItemParameters = "0=0,1=0,2=1,3=1", jobParameters = "parameter")
@Slf4j
public class MultipleElasticJob implements SimpleJob {

   @Override
   public void execute(ShardingContext shardingContext) {
      log.info("Thread ID: {}, ShardingTotalCount: {}, ShardingItem: {}, ShardingParameter: {}, JobName: {}, JobParameter: {}",
              Thread.currentThread().getId(),
              shardingContext.getShardingTotalCount(),
              shardingContext.getShardingItem(),
              shardingContext.getShardingParameter(),
              shardingContext.getJobName(),
              shardingContext.getJobParameter()
      );
   }
}
  1. 使用ElasticJobService,可以手工增加,删除,触发job
public interface ElasticJobService {

    /**
     * 添加简单作业
     *
     * @param elasticJob
     * @param jobName
     * @param shardingTotalCount
     * @param cron
     */
    JobBootstrap addJob(ElasticJob elasticJob, String jobName, int shardingTotalCount, String cron);

    /**
     * 添加简单作业
     *
     * @param elasticJob
     * @param jobName
     * @param shardingTotalCount
     * @param cron
     * @param jobParameter
     * @param shardingItemParameters
     */
    JobBootstrap addJob(ElasticJob elasticJob, String jobName, int shardingTotalCount, String cron, String jobParameter, String... shardingItemParameters);

    /**
     * 添加简单作业
     *
     * @param jobConfiguration
     * @param jobBootstrapBeanName 非必填
     * @param elasticJob
     * @return
     */
    JobBootstrap addJob(JobConfiguration jobConfiguration, String jobBootstrapBeanName, ElasticJob elasticJob);

    /**
     * 添加数据流作业
     *
     * @param jobConfiguration
     * @param jobBootstrapBeanName
     * @param elasticJob
     * @param streamingProcess     是否开启流式处理 默认false
     * @return
     */
    <T> JobBootstrap addDataFlowJob(JobConfiguration jobConfiguration, String jobBootstrapBeanName, DataflowJob<T> elasticJob, boolean streamingProcess);

    /**
     * 添加定时Script调度
     *
     * @param jobConfiguration
     * @param scriptCommendLine
     */
    JobBootstrap addScriptJob(JobConfiguration jobConfiguration, String jobBootstrapBeanName, String scriptCommendLine);

    /**
     * 添加定时Http调度
     *
     * @param jobConfiguration
     * @param httpProp
     */
    JobBootstrap addHttpJob(JobConfiguration jobConfiguration, String jobBootstrapBeanName, HttpJobProp httpProp);

    /**
     * 移除job
     *
     * @param jobName
     */
    void removeJob(String jobName);

    /**
     * 移除job
     *
     * @param jobName
     * @param jobBootstrapBeanName
     */
    void removeJob(String jobName, String jobBootstrapBeanName);

    /**
     * get job configuration.
     *
     * @param jobName job name
     * @return job configuration
     */
    JobConfigurationPOJO getJobConfiguration(String jobName);

    /**
     * Update job configuration.
     *
     * @param jobConfig job configuration
     */
    void updateJobConfiguration(JobConfigurationPOJO jobConfig);

    /**
     * Remove job configuration.
     *
     * @param jobName job name
     */
    void removeJobConfiguration(String jobName);

    /**
     * Trigger job to run at once.
     *
     * <p>Job will not start until it does not conflict with the last running job, and this tag will be automatically cleaned up after it starts.</p>
     *
     * @param jobName job name
     */
    void trigger(String jobName);

    /**
     * Get jobs total count.
     *
     * @return jobs total count.
     */
    int getJobsTotalCount();

    /**
     * Get all jobs brief info.
     *
     * @return all jobs brief info.
     */
    Collection<JobBriefInfo> getAllJobsBriefInfo();

    /**
     * Get job brief info.
     *
     * @param jobName job name
     * @return job brief info
     */
    JobBriefInfo getJobBriefInfo(String jobName);
}
  1. 更多配置
#配置策略,自定义策略参考官方文档定义后配置到这里即可
#错误策略.  LOG,THROW,IGNORE,EMAIL,WECHAT,DINGTALK
elasticjob.job-error-handler-type=LOG
#分片策略. AVG_ALLOCATION,ODEVITY,ROUND_ROBIN
elasticjob.job-sharding-strategy-type=AVG_ALLOCATION
#线程策略. CPU,SINGLE_THREAD
elasticjob.job-executor-service-handler-type=CPU

#追踪
elasticjob.tracing.type=RDB
#multiple datasource
elasticjob.tracing.data-source-bean-name=db1MasterSlaveRoutingDatasource

#配置邮件提醒策略属性
elasticjob.props.email.host=host
elasticjob.props.email.port=465
elasticjob.props.email.username=username
elasticjob.props.email.password=password
elasticjob.props.email.useSsl=true
elasticjob.props.email.subject=ElasticJob error message
elasticjob.props.email.from=from@xxx.xx
elasticjob.props.email.to=to1@xxx.xx,to2@xxx.xx
elasticjob.props.email.cc=cc@xxx.xx
elasticjob.props.email.bcc=bcc@xxx.xx
elasticjob.props.email.debug=false

#配置微信提醒策略属性
elasticjob.props.wechat.webhook=you_webhook
elasticjob.props.wechat.connectTimeout=3000
elasticjob.props.wechat.readTimeout=5000

#配置钉钉提醒策略属性
elasticjob.props.dingtalk.webhook=you_webhook
elasticjob.props.dingtalk.keyword=you_keyword
elasticjob.props.dingtalk.secret=you_secret
elasticjob.props.dingtalk.connectTimeout=3000
elasticjob.props.dingtalk.readTimeout=5000
最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
平台声明:文章内容(如有图片或视频亦包括在内)由作者上传并发布,文章内容仅代表作者本人观点,简书系信息发布平台,仅提供信息存储服务。

推荐阅读更多精彩内容