elastic-job 分佈式定時任務框架在 SpringBoot 中如何使用（一）初始化任務並定時執行

時間 2019-12-11

標籤 elastic job 分佈式定時任務框架 springboot 如何使用初始化執行欄目系統架構简体版

原文原文鏈接

第一篇須要實現一個最簡單的需求：某個任務定時執行，多臺機子只讓其中一臺機子執行任務html

1、安裝分佈式應用程序協調服務 zookeeper，安裝步驟在連接裏面java

　Linux(Centos7)下安裝 zookeeper docker版集羣　git

2、在springboot項目中引入 elastic-job 依賴，我這裏用的 springboot 2.0.5 版本github

整合代碼參考官方的springboot demo spring

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

對應的 elastic-job 版本express

<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 -->

依賴XML代碼

application.yaml 加入配置apache

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

application.yaml 配置代碼

3、做業代碼（simpleJob類型的任務），把官方的demo拿到本身的項目中來就行，注意添加一個@Component註解在類上api

/*
 * 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());
        }
    }
}

做業任務代碼

4、配置做業（直接將官方的配置代碼拿進本身的項目中）springboot

/*
 * 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();
    }
}