elastic-job 分布式定时任务框架 在 SpringBoot 中如何使用（一）初始化任务并定时执行

第一篇需要实现一个最简单的需求：某个任务定时执行，多台机子只让其中一台机子执行任务

 

一、安装 分布式应用程序协调服务 zookeeper，安装步骤在链接里面

 　Linux(Centos7)下安装 zookeeper docker版 集群　

 

二、在springboot项目中引入 elastic-job 依赖，我这里用的 springboot 2.0.5 版本

整合代码参考官方的springboot demo 

https://github.com/elasticjob/elastic-job-example

 

 对应的  elastic-job 版本

<elastic-job.version>2.1.5</elastic-job.version>

        <!-- apache elastic-job start -->
        <dependency>
            <groupId>com.dangdang</groupId>
            <artifactId>elastic-job-lite-core</artifactId>
            <version>${elastic-job.version}</version>
        </dependency>
        <dependency>
            <groupId>com.dangdang</groupId>
            <artifactId>elastic-job-lite-spring</artifactId>
            <version>${elastic-job.version}</version>
        </dependency>
        <!-- apache elastic-job end -->

 

application.yaml 加入配置

regCenter:
  serverList: localhost:2181
  namespace: my-project

simpleJob:
  cron: 0/15 * * * * ?
  shardingTotalCount: 1
  shardingItemParameters: 0=a

dataflowJob:
  cron: 0/5 * * * * ?
  shardingTotalCount: 3
  shardingItemParameters: 0=Beijing,1=Shanghai,2=Guangzhou

 

三、作业代码（simpleJob类型的任务），把官方的demo拿到自己的项目中来就行，注意添加一个@Component注解在类上

 

/*
 * Copyright 1999-2015 dangdang.com.
 * <p>
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 * </p>
 */

package com.dianji.task_server.job.exec;

import com.dangdang.ddframe.job.api.ShardingContext;
import com.dangdang.ddframe.job.api.simple.SimpleJob;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.text.MessageFormat;
import java.text.SimpleDateFormat;
import java.util.Date;

@Slf4j
@Component
public class SpringSimpleJob implements SimpleJob {
    @Value("${flag}")
    private String serverFlag;

    @Override
    public void execute(final ShardingContext shardingContext) {
        int shardingItem = shardingContext.getShardingItem();
        if (shardingItem == 0) {
            log.info("这是来自『{}』的任务，SIMPLE「SpringSimpleJob_execute」", serverFlag + ":" +
                    shardingContext.getShardingParameter());
            String logStr = "Item: {0} | Parameter: {1} | : {2} | Time: {3} | Thread: {4} | {5}";
            logStr = MessageFormat.format(logStr, shardingContext.getShardingItem(),
                    shardingContext.getShardingParameter(),
                    new SimpleDateFormat("HH:mm:ss").format(new Date()),
                    Thread.currentThread().getId(), "SIMPLE「SpringSimpleJob_execute」");
            log.info(logStr);
        } else if (shardingItem == 1) {
            log.info("这是来自『{}』的任务", serverFlag + ":" + shardingContext.getShardingParameter());
        } else if (shardingItem == 2) {
            log.info("这是来自『{}』的任务", shardingContext.getShardingParameter());
        }
    }
}

 

四、配置作业（直接将官方的配置代码拿进自己的项目中）

/*
 * Copyright 1999-2015 dangdang.com.
 * <p>
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 * </p>
 */

package com.dianji.task_server.job.config;

import com.dangdang.ddframe.job.api.simple.SimpleJob;
import com.dangdang.ddframe.job.config.JobCoreConfiguration;
import com.dangdang.ddframe.job.config.simple.SimpleJobConfiguration;
import com.dangdang.ddframe.job.event.JobEventConfiguration;
import com.dangdang.ddframe.job.lite.api.JobScheduler;
import com.dangdang.ddframe.job.lite.config.LiteJobConfiguration;
import com.dangdang.ddframe.job.lite.spring.api.SpringJobScheduler;
import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperRegistryCenter;
import com.dianji.task_server.job.exec.SpringSimpleJob;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.Resource;

@Configuration
public class SimpleJobConfig {

    @Resource
    private ZookeeperRegistryCenter regCenter;

    @Resource
    private JobEventConfiguration jobEventConfiguration;

    @Bean
    public SimpleJob simpleJob() {
        return new SpringSimpleJob();
    }

    @Bean(initMethod = "init")
    public JobScheduler simpleJobScheduler(final SimpleJob simpleJob, @Value("${simpleJob.cron}") final String cron, @Value("${simpleJob.shardingTotalCount}") final int shardingTotalCount,
                                           @Value("${simpleJob.shardingItemParameters}") final String shardingItemParameters) {
        return new SpringJobScheduler(simpleJob, regCenter, getLiteJobConfiguration(simpleJob.getClass(), cron, shardingTotalCount, shardingItemParameters), jobEventConfiguration);
    }

    private LiteJobConfiguration getLiteJobConfiguration(final Class<? extends SimpleJob> jobClass, final String cron, final int shardingTotalCount, final String shardingItemParameters) {
        return LiteJobConfiguration.newBuilder(new SimpleJobConfiguration(JobCoreConfiguration.newBuilder(
                jobClass.getName(), cron, shardingTotalCount).shardingItemParameters(shardingItemParameters).build(), jobClass.getCanonicalName())).overwrite(true).build();
    }
}

 

五、最简单的整合步骤就完成了，启动应用（分别启动三个进程），看见任务开始执行了

最简单的任务需求得到了满足，定时执行，多台机子只让其中一台机子执行任务

 

 

关于分片的数量设置，实际操作了一把

 

假如：有3个进程实例

并且elastic-job 设置  

分片 1 时：3个进程只有1一个进程执行1次任务，共执行1次任务

分片 2 时：3个进程中的2个进程一共执行2次任务，共执行2次任务

分片 3 时：3个进程每个执行1次任务，共执行3次任务

分片 4 时：3个进程同时一共执行4次任务，共执行4次任务

分片 5 时：3个进程同时一共执行5次任务，共执行5次任务

 

分片为X时，此任务同时在Y台个进程里面执行X次。

 

这样就可以根据不同的使用场景配置不同的分片数量了