方案一：使用異步任務(wù)+定時(shí)任務(wù)可以結(jié)合使用，實(shí)現(xiàn)在指定時(shí)間或周期內(nèi)異步執(zhí)行任務(wù)的功能（單單使用定時(shí)任務(wù)會(huì)阻塞主線程）

適用場(chǎng)景：
Spring 定時(shí)任務(wù)是 Spring 框架提供的一種輕量級(jí)的任務(wù)調(diào)度方案，它的特點(diǎn)是簡(jiǎn)單易用、輕量級(jí)。Spring 定時(shí)任務(wù)的執(zhí)行是在單個(gè)節(jié)點(diǎn)上進(jìn)行的，如果需要分布式任務(wù)調(diào)度，需要自己實(shí)現(xiàn)相應(yīng)的解決方案。

使用步驟

1.導(dǎo)入依賴(lài)版本自己控制

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter</artifactId>
    </dependency>
    
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-task</artifactId>
    </dependency>
</dependencies>

2.啟動(dòng)類(lèi)加上@EnableScheduling

3.編寫(xiě)業(yè)務(wù)
cron表達(dá)式在線生成
https://cron.qqe2.com/

import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
/**
 * 定時(shí)任務(wù)
 *      1、@EnableScheduling 開(kāi)啟定時(shí)任務(wù)
 *      2、@Scheduled  開(kāi)啟一個(gè)定時(shí)任務(wù)
 *      3、自動(dòng)配置類(lèi) TaskSchedulingAutoConfiguration
 *
 * 異步任務(wù)
 *      1、@EnableAsync 開(kāi)啟異步任務(wù)功能
 *      2、@Async 給希望異步執(zhí)行的方法上標(biāo)注
 *      3、自動(dòng)配置類(lèi) TaskExecutionAutoConfiguration 屬性綁定在TaskExecutionProperties
 *
 */
@Slf4j
@Component
public class HelloSchedule {
    /**
     * 1、Spring中6位組成，不允許第7位的年
     * 2、在周幾的位置，1-7代表周一到周日； MON-SUN
     * 3、定時(shí)任務(wù)不應(yīng)該阻塞。默認(rèn)是阻塞的
     *      1）、可以讓業(yè)務(wù)運(yùn)行以異步的方式，自己提交到線程池
     *              CompletableFuture.runAsync(()->{
     *                  xxxxService.hello();
     *              },executor);
     *      2）、支持定時(shí)任務(wù)線程池；設(shè)置 TaskSchedulingProperties；
     *              spring.task.scheduling.pool.size=5
     *      3）、讓定時(shí)任務(wù)異步執(zhí)行
     *          異步任務(wù)；
     *     解決：使用異步+定時(shí)任務(wù)來(lái)完成定時(shí)任務(wù)不阻塞的功能；

    @Async
    @Scheduled(cron = "* * * ? * 5")//表達(dá)式
    public void hello()  {
        log.info("執(zhí)行業(yè)務(wù)...");
    }
}

方案二：引入中間件xxl-job

XXL-Job 是一個(gè)基于 Spring 的分布式任務(wù)調(diào)度框架，它的特點(diǎn)是分布式、可靠性高、易于管理。XXL-Job 提供了任務(wù)調(diào)度、執(zhí)行、報(bào)警、日志等完整的功能，支持任務(wù)失敗重試、任務(wù)失敗告警等機(jī)制，能夠保證任務(wù)執(zhí)行的高可靠性和穩(wěn)定性。同時(shí)，XXL-Job 的任務(wù)執(zhí)行是分布式的，可以將任務(wù)分發(fā)到多個(gè)節(jié)點(diǎn)執(zhí)行，提高任務(wù)執(zhí)行效率。

使用 XXL-Job 需要在項(xiàng)目中引入 XXL-Job 的相關(guān)依賴(lài)，并進(jìn)行相關(guān)配置。同時(shí)，XXL-Job 還提供了一個(gè)可視化的任務(wù)管理平臺(tái)，可以方便地管理和監(jiān)控任務(wù)的執(zhí)行情況。

XXL-Job 的使用場(chǎng)景一般是執(zhí)行復(fù)雜的分布式定時(shí)任務(wù)，例如：訂單超時(shí)未支付處理、大數(shù)據(jù)分析等。如果只是執(zhí)行簡(jiǎn)單的定時(shí)任務(wù)，則可以使用 Spring 定時(shí)任務(wù)。

使用步驟

A.本地安裝
碼云：https://gitee.com/xuxueli0323/xxl-job
1.下載項(xiàng)目 2.導(dǎo)入里面的數(shù)據(jù)庫(kù) 3.啟動(dòng)服務(wù)
B.虛擬機(jī)安裝
要在 Linux 系統(tǒng)上使用 Docker 部署 XXL-Job，可以按照以下步驟進(jìn)行操作：
1.拉取 XXL-Job 鏡像：可以從 Docker Hub 上拉取 XXL-Job 鏡像，執(zhí)行以下命令：
docker pull xuxueli/xxl-job-admin:latest # 拉取 XXL-Job Admin 鏡像
docker pull xuxueli/xxl-job-executor:latest # 拉取 XXL-Job Executor 鏡像
2.創(chuàng)建 XXL-Job 容器：

a. 創(chuàng)建 XXL-Job Admin 容器：
docker run -d --name xxl-job-admin -p 8080:8080 xuxueli/xxl-job-admin
這個(gè)命令將會(huì)創(chuàng)建一個(gè)名為 xxl-job-admin 的容器，并且將容器的 8080 端口映射到宿主機(jī)的 8080 端口，使得我們可以通過(guò)瀏覽器訪問(wèn)到 XXL-Job 的管理界面。
b. 創(chuàng)建 XXL-Job Executor 容器：
docker run -d --name xxl-job-executor
-e PARAMS=“–xxl.job.admin.address=http://xxl-job-admin:8080/xxl-job-admin”
xuxueli/xxl-job-executor
這個(gè)命令將會(huì)創(chuàng)建一個(gè)名為 xxl-job-executor 的容器，并且將 XXL-Job Admin 的地址配置到容器中，使得 Executor 能夠與 Admin 進(jìn)行通信。
至此，XXL-Job 在 Docker 中的部署就完成了。通過(guò)瀏覽器訪問(wèn) http://ip:8080/xxl-job-admin/，即可進(jìn)入 XXL-Job 的管理界面。

那么如何使用xxl-job進(jìn)行任務(wù)調(diào)度呢
1.進(jìn)入管理界面，新增執(zhí)行器

2.在需要添加任務(wù)調(diào)度的服務(wù)中添加依賴(lài)

    <dependency>
        <groupId>com.xuxueli</groupId>
        <artifactId>xxl-job-core</artifactId>
    </dependency>

3.配置yml，改ip和hander

xxl:
  job:
    admin: 
      addresses: http://----ip---/xxl-job-admin
    executor:
      appname: #你在管理頁(yè)面命名的hander
      address: 
      ip: 
      port: 9999
      logpath: /data/applogs/xxl-job/jobhandler
      logretentiondays: 30
    accessToken: default_token

4.復(fù)制配置文件，文件在xxl-job項(xiàng)目的 xxl-job-2.3.1\xxl-job-executor-samples\xxl-job-executor-sample-springboot\src\main\java\com\xxl\job\executor\core\config 的XxlJobConfig

package com.xuecheng.media.config;


import com.xxl.job.core.executor.impl.XxlJobSpringExecutor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * xxl-job config
 *
 * @author xuxueli 2017-04-28
 */
@Configuration
public class XxlJobConfig {
    private Logger logger = LoggerFactory.getLogger(XxlJobConfig.class);

    @Value("${xxl.job.admin.addresses}")
    private String adminAddresses;

    @Value("${xxl.job.accessToken}")
    private String accessToken;

    @Value("${xxl.job.executor.appname}")
    private String appname;

    @Value("${xxl.job.executor.address}")
    private String address;

    @Value("${xxl.job.executor.ip}")
    private String ip;

    @Value("${xxl.job.executor.port}")
    private int port;

    @Value("${xxl.job.executor.logpath}")
    private String logPath;

    @Value("${xxl.job.executor.logretentiondays}")
    private int logRetentionDays;


    @Bean
    public XxlJobSpringExecutor xxlJobExecutor() {
        logger.info(">>>>>>>>>>> xxl-job config init.");
        XxlJobSpringExecutor xxlJobSpringExecutor = new XxlJobSpringExecutor();
        xxlJobSpringExecutor.setAdminAddresses(adminAddresses);
        xxlJobSpringExecutor.setAppname(appname);
        xxlJobSpringExecutor.setAddress(address);
        xxlJobSpringExecutor.setIp(ip);
        xxlJobSpringExecutor.setPort(port);
        xxlJobSpringExecutor.setAccessToken(accessToken);
        xxlJobSpringExecutor.setLogPath(logPath);
        xxlJobSpringExecutor.setLogRetentionDays(logRetentionDays);

        return xxlJobSpringExecutor;
    }

    /**
     * 針對(duì)多網(wǎng)卡、容器內(nèi)部署等情況，可借助 "spring-cloud-commons" 提供的 "InetUtils" 組件靈活定制注冊(cè)IP；
     *
     *      1、引入依賴(lài)：
     *          <dependency>
     *             <groupId>org.springframework.cloud</groupId>
     *             <artifactId>spring-cloud-commons</artifactId>
     *             <version>${version}</version>
     *         </dependency>
     *
     *      2、配置文件，或者容器啟動(dòng)變量
     *          spring.cloud.inetutils.preferred-networks: 'xxx.xxx.xxx.'
     *
     *      3、獲取IP
     *          String ip_ = inetUtils.findFirstNonLoopbackHostInfo().getIpAddress();
     */


}

成功的標(biāo)志

配置已經(jīng)完成，那么如何使用呢？其實(shí)使用跟spring自帶的定時(shí)任務(wù)很像
1.在服務(wù)中定義任務(wù)，一般寫(xiě)在service層

2.登上x(chóng)xl-job的admin配置cron和spring的定時(shí)任務(wù)是一樣的，但是提供了更人性化，和更豐富的功能.

自己自定義配置任務(wù)的執(zhí)行周期和策略

方案三：elastic-job
elastic-job和xxl-job都是適合使用分布式任務(wù)調(diào)用的框架，兩者的比較可以自行搜索，在這里只解釋用法
1.下載zookeeper，elastic-job是以zk為注冊(cè)中心，Elastic-Job 使用 ZooKeeper 來(lái)實(shí)現(xiàn)分布式任務(wù)的協(xié)調(diào)和調(diào)度機(jī)制，通過(guò) ZooKeeper 來(lái)管理任務(wù)的運(yùn)行狀態(tài)、分片策略、任務(wù)節(jié)點(diǎn)等信息，從而實(shí)現(xiàn)分布式任務(wù)的動(dòng)態(tài)調(diào)度和高可用性。因此，在使用 Elastic-Job 之前，需要先部署和配置好 ZooKeeper。
這里講解普通下載不是docker，docker玩法可自學(xué)，也可直接下載到本地
1.1.上傳,將zookeeper-3.4.11.tar.gz上傳到/usr/local/software目錄下

1.2.解壓文件到指定目錄

tar -zxvf /usr/local/software/zookeeper-3.4.11.tar.gz -C /usr/local/

1.3.拷貝配置文件

cp /usr/local/zookeeper-3.4.11/conf/zoo_sample.cfg /usr/local/zookeeper-3.4.11/conf/zoo.cfg

1.4.啟動(dòng)

/usr/local/zookeeper-3.4.11/bin/zkServer.sh start

1.5.檢查進(jìn)程是否開(kāi)啟

jps

需要查看到QuorumPeerMain進(jìn)程

2.導(dǎo)入依賴(lài)

        <dependency>
            <groupId>com.dangdang</groupId>
            <artifactId>elastic-job-lite-spring</artifactId>
            <version>2.1.5</version>
        </dependency>
<!--     job和zk連接的工具包-->
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-recipes</artifactId>
            <version>2.10.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-framework</artifactId>
            <version>2.10.0</version>
        </dependency>
        <dependency>
            <groupId>com.google.guava</groupId>
            <artifactId>guava</artifactId>
            <version>18.0</version>
        </dependency>

3.配置yml

elasticjob:
  zookeeper-url: 192.168.200.128:2181
  group-name: shop-job-group
jobCron:
  initSeckillProduct: 0 0/1 * * * ?
  userCache: 0 0/1 * * * ?

4.導(dǎo)入工具類(lèi)

mport com.dangdang.ddframe.job.api.simple.SimpleJob;
import com.dangdang.ddframe.job.config.JobCoreConfiguration;
import com.dangdang.ddframe.job.config.JobTypeConfiguration;
import com.dangdang.ddframe.job.config.dataflow.DataflowJobConfiguration;
import com.dangdang.ddframe.job.config.simple.SimpleJobConfiguration;
import com.dangdang.ddframe.job.lite.config.LiteJobConfiguration;
import org.apache.commons.lang.StringUtils;

/**
 * Created by lanxw
 * 工具類(lèi)，用于創(chuàng)建LiteJobConfiguration對(duì)象
 */
public class ElasticJobUtil {
    /**
     * 創(chuàng)建作業(yè)核心配置信息和Lite作業(yè)根配置信息（用于分片執(zhí)行任務(wù)）
     *
     * @param jobClass               作業(yè)實(shí)現(xiàn)類(lèi)
     * @param cron                   作業(yè)觸發(fā)時(shí)間表達(dá)式
     * @param shardingTotalCount     分片總數(shù)
     * @param shardingItemParameters 分片參數(shù),按照什么進(jìn)行分片，
     *                               如有3個(gè)分片 且 shardingItemParameters="0=10,1=12,2=14"，
     *                               那么會(huì)有三個(gè)線程執(zhí)行execute方法 調(diào)用shardingContext.getShardingParameter() 分別對(duì)應(yīng)著 10 12 14
     * @param dataflowType           是否為數(shù)據(jù)流作業(yè)類(lèi)型
     * @return Lite作業(yè)配置信息
     */
    public static LiteJobConfiguration createJobConfiguration(final Class<? extends SimpleJob> jobClass,
                                                              final String cron,
                                                              final int shardingTotalCount,
                                                              final String shardingItemParameters,
                                                              boolean dataflowType) {
        // 定義作業(yè)核心配置
        JobCoreConfiguration.Builder jobCoreConfigurationBuilder = JobCoreConfiguration.newBuilder(jobClass.getSimpleName(), cron, shardingTotalCount);
        if(!StringUtils.isEmpty(shardingItemParameters)){
            jobCoreConfigurationBuilder.shardingItemParameters(shardingItemParameters);
        }
        JobTypeConfiguration jobConfig = null;
        if(dataflowType){
            // 創(chuàng)建數(shù)據(jù)流作業(yè)配置
            jobConfig = new DataflowJobConfiguration(jobCoreConfigurationBuilder.build(),jobClass.getCanonicalName(),true);
        }else {
            // 創(chuàng)建簡(jiǎn)單作業(yè)配置
            jobConfig = new SimpleJobConfiguration(jobCoreConfigurationBuilder.build(), jobClass.getCanonicalName());
        }
        // 定義Lite作業(yè)根配置，允許重寫(xiě)覆蓋方法的配置
        LiteJobConfiguration simpleJobRootConfig = LiteJobConfiguration.newBuilder(jobConfig).overwrite(true).build();
        return simpleJobRootConfig;
    }

    /**
     * 創(chuàng)建默認(rèn)的簡(jiǎn)單作業(yè)配置信息（執(zhí)行簡(jiǎn)單的任務(wù)）
     *
     * @param jobClass 作業(yè)實(shí)現(xiàn)類(lèi)
     * @param cron     作業(yè)觸發(fā)時(shí)間表達(dá)式
     * @return Lite作業(yè)配置信息
     */
    public static LiteJobConfiguration createDefaultSimpleJobConfiguration(final Class<? extends SimpleJob> jobClass, final String cron) {
        return createJobConfiguration(jobClass,cron,1,null,false);
    }

    /**
     * 創(chuàng)建默認(rèn)的數(shù)據(jù)流作業(yè)配置信息
     *
     * @param jobClass 作業(yè)實(shí)現(xiàn)類(lèi)
     * @param cron     作業(yè)觸發(fā)時(shí)間表達(dá)式
     * @return Lite作業(yè)配置信息
     */
    public static LiteJobConfiguration createDefaultDataFlowJobConfiguration(final Class<? extends SimpleJob> jobClass, final String cron) {
        return createJobConfiguration(jobClass,cron,1,null,true);
    }
}

5.配置注冊(cè)中心zk

/**
 * Created by lanxw
 * 注冊(cè)中心配置對(duì)象ZK
 */
@Configuration
public class RegistryCenterConfig {
    @Bean(initMethod = "init")
    public CoordinatorRegistryCenter createRegistryCenter(@Value("${elasticjob.zookeeper-url}") String zookeeperUrl, @Value("${elasticjob.group-name}") String groupName) {
        //zk的配置
        ZookeeperConfiguration zookeeperConfiguration = new ZookeeperConfiguration(zookeeperUrl,groupName);
        //設(shè)置zk超時(shí)時(shí)間
        zookeeperConfiguration.setSessionTimeoutMilliseconds(100);
        //創(chuàng)建注冊(cè)中心
        CoordinatorRegistryCenter zookeeperRegistryCenter = new ZookeeperRegistryCenter(zookeeperConfiguration);
        return zookeeperRegistryCenter;
    }
}

6.寫(xiě)任務(wù)類(lèi)，execute就是任務(wù)執(zhí)行邏輯

/**
 * 商品上架的定時(shí)任務(wù)
 */
@Component
@Setter
@Getter
@RefreshScope//該 Bean 的配置可以在運(yùn)行時(shí)被動(dòng)態(tài)刷新，即可以從配置中心獲取最新的配置值。
@Slf4j
public class InitSeckillProductJob implements SimpleJob {
    @Value("${jobCron.initSeckillProduct}")
    private String cron;

    @Autowired
    private SeckillProductFeignAPI seckillProductFeignAPI;

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @Override
    public void execute(ShardingContext shardingContext) {
        log.info("秒殺商品上架任務(wù)正在執(zhí)行。。。");
        //遠(yuǎn)程調(diào)用秒殺服務(wù)
        String time = shardingContext.getShardingParameter();
        Result<List<SeckillProductVo>> listResult = seckillProductFeignAPI.queryByTimeForJob(Integer.parseInt(time));
        if (listResult==null||listResult.hasError()){
            //通知管理員，秒殺服務(wù)有問(wèn)題
            return;
        }
        List<SeckillProductVo> data = listResult.getData();
        //刪除redis里的舊秒殺數(shù)據(jù)
        String key= JobRedisKey.SECKILL_PRODUCT_HASH.getRealKey(time);
        stringRedisTemplate.delete(key);
        for (SeckillProductVo vo : data) {
            stringRedisTemplate.opsForHash().put(key,String.valueOf(vo.getId()), JSON.toJSONString(vo));
        }
        log.info("秒殺商品上架任務(wù)執(zhí)行完成。。。");
    }
}

7.任務(wù)注冊(cè)和分片指定的邏輯等，可以到官網(wǎng)查看用法

@Configuration
public class BusinessJobConfig {
    @Bean(initMethod = "init")
    public SpringJobScheduler initUserCacheJob(CoordinatorRegistryCenter registryCenter, UserCacheJob userCacheJob){
        LiteJobConfiguration jobConfiguration = ElasticJobUtil.createDefaultSimpleJobConfiguration(userCacheJob.getClass(), userCacheJob.getCron());
        SpringJobScheduler springJobScheduler = new SpringJobScheduler(userCacheJob, registryCenter,jobConfiguration );
        return springJobScheduler;
    }

    @Bean(initMethod = "init")
    public SpringJobScheduler initSPJob(CoordinatorRegistryCenter registryCenter, InitSeckillProductJob initSeckillProductJob){
        //創(chuàng)建配置
        LiteJobConfiguration jobConfiguration = ElasticJobUtil.createJobConfiguration(
                initSeckillProductJob.getClass(),
                initSeckillProductJob.getCron(),
                3, "0=10,1=12,2=14",false);
        SpringJobScheduler springJobScheduler = new SpringJobScheduler(initSeckillProductJob, registryCenter,jobConfiguration );
        return springJobScheduler;
    }
}

這樣任務(wù)即可根據(jù)您的配置文件yml的cron表達(dá)式執(zhí)行代碼文章來(lái)源地址http://www.zghlxwxcb.cn/news/detail-412704.html

到了這里，關(guān)于分布式任務(wù)調(diào)度，定時(shí)任務(wù)的處理方案的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！