定時(shí)任務(wù)場景

在開發(fā)過程中，根據(jù)需求和業(yè)務(wù)的不同經(jīng)常會(huì)有很多場景需要用到不同特性的定時(shí)任務(wù)，針對(duì)這些場景，這里提供不同的一個(gè)實(shí)現(xiàn)思路。定時(shí)任務(wù)可能需要的特性如下：

• 多線程執(zhí)行：即一個(gè)定時(shí)任務(wù)是需要多線程去跑的，因?yàn)橐粋€(gè)線程太慢了
• 分布式執(zhí)行：在多線程的基礎(chǔ)上，用多臺(tái)機(jī)器的算力去執(zhí)行一個(gè)定時(shí)任務(wù)
• 動(dòng)態(tài)時(shí)間的定時(shí)任務(wù)：即定時(shí)任務(wù)的開始時(shí)間是不確定的
• 連續(xù)上下文多線程定時(shí)任務(wù)：在使用多線程執(zhí)行定時(shí)任務(wù)的時(shí)候，嚴(yán)格按照任務(wù)的順序來執(zhí)行，即任務(wù)ABC,多線程執(zhí)行完A,然后再執(zhí)行B和C,不能存在A還有沒跑完的線程，B已經(jīng)開始的情況。
• 可暫停繼續(xù)的定時(shí)任務(wù)：即一個(gè)任務(wù)執(zhí)行了一半，可以暫停后繼續(xù)運(yùn)行。

針對(duì)以上的幾種特性，這里講一講相對(duì)簡單的實(shí)現(xiàn)方式

不同定時(shí)任務(wù)特性對(duì)應(yīng)的實(shí)現(xiàn)方式

多線程執(zhí)行

多線程執(zhí)行相對(duì)來說是個(gè)比較簡單的需求，只需要定時(shí)任務(wù)觸發(fā)的時(shí)候，使用線程池去執(zhí)行任務(wù)，在springboot中，只需要使用@Async注解就行了。

//定時(shí)任務(wù)代碼
@Scheduled(fixedDelay = 60000)
    public void doSomeThing() {
        servive.doA();
    }
-------------------------------------------
//執(zhí)行定時(shí)任務(wù)邏輯代碼
@Service
public class Service(){
    @Async("A-config")
    public void doA(){
        //這里執(zhí)行邏輯省略
    }
}
-------------------------------------------
@Bean("updateById")
    public Executor updateById() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        // 核心線程數(shù)：線程池創(chuàng)建時(shí)候初始化的線程數(shù)
        executor.setCorePoolSize(64);
        // 最大線程數(shù)：線程池最大的線程數(shù)，只有在緩沖隊(duì)列滿了之后才會(huì)申請超過核心線程數(shù)的線程
        executor.setMaxPoolSize(128);
        // 緩沖隊(duì)列：用來緩沖執(zhí)行任務(wù)的隊(duì)列
        executor.setQueueCapacity(500);
        // 允許線程的空閑時(shí)間60秒：當(dāng)超過了核心線程之外的線程在空閑時(shí)間到達(dá)之后會(huì)被銷毀
        executor.setKeepAliveSeconds(60);
        // 線程池名的前綴：設(shè)置好了之后可以方便我們定位處理任務(wù)所在的線程池
        executor.setThreadNamePrefix("do-somthing-");
        // 緩沖隊(duì)列滿了之后的拒絕策略：這里調(diào)用主線程執(zhí)行
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.setWaitForTasksToCompleteOnShutdown(true);
        executor.initialize();
        return executor;
    }

分布式執(zhí)行

一般分布式定時(shí)任務(wù)主要是兩個(gè)問題，一個(gè)是觸發(fā)，一個(gè)是分發(fā)。觸發(fā)需要考慮到只能有一個(gè)觸發(fā)成功，需要根據(jù)定時(shí)任務(wù)的間隔，指定合理的競爭策略，比如每日定時(shí)任務(wù)，那就用當(dāng)日凌晨當(dāng)key，利用redis的setnx原子操作來競爭。 然后就是分發(fā)問題，一般任務(wù)觸發(fā)后，利用多臺(tái)機(jī)分布式執(zhí)行定時(shí)任務(wù)，由一臺(tái)機(jī)觸發(fā)后，把任務(wù)分發(fā)到中間件中，然后多臺(tái)機(jī)器的服務(wù)去消費(fèi)這些任務(wù)，這個(gè)中間件一般是隊(duì)列或者redis或者數(shù)據(jù)庫。

動(dòng)態(tài)時(shí)間的定時(shí)任務(wù)

當(dāng)需要在一個(gè)不確定的時(shí)間去執(zhí)行任務(wù)，比如我需要定時(shí)任務(wù)開啟一場折扣活動(dòng)，而這個(gè)開啟時(shí)間取決于業(yè)務(wù)的其他條件出發(fā)，這個(gè)開始時(shí)間會(huì)在某一個(gè)業(yè)務(wù)時(shí)刻存入數(shù)據(jù)庫或者reids，那就需要一個(gè)任務(wù)觸發(fā)檢測器了，可以跑一個(gè)1s為間隔的定時(shí)任務(wù)，每秒鐘都檢查一次這個(gè)任務(wù)是否需要執(zhí)行或者是否到了執(zhí)行時(shí)間，如果需要執(zhí)行就立刻執(zhí)行相應(yīng)的任務(wù)邏輯。

連續(xù)上下文多線程定時(shí)任務(wù)

實(shí)際業(yè)務(wù)場景經(jīng)常會(huì)有這種情況，我需要連續(xù)的執(zhí)行A->B->C這三個(gè)任務(wù)，一定要按嚴(yán)格的順序執(zhí)行，如果我給這三個(gè)任務(wù)分別設(shè)置三個(gè)執(zhí)行時(shí)間，那后續(xù)任務(wù)如果超時(shí)或者執(zhí)行時(shí)間超過預(yù)期，就有可能出現(xiàn)B任務(wù)在A沒結(jié)束就執(zhí)行。所以只能按同一個(gè)定時(shí)任務(wù)觸發(fā)來執(zhí)行。而這時(shí)候我又需要多線程來提高效率。所以這個(gè)場景的重點(diǎn)在于，子任務(wù)結(jié)束的判定也就是線程執(zhí)行完畢的判斷。這里有種很簡單的實(shí)現(xiàn)方式。

實(shí)現(xiàn)這個(gè)場景的基本思路就是，在執(zhí)行任務(wù)A時(shí)候創(chuàng)建線程池去執(zhí)行A任務(wù)邏輯，然后分發(fā)完之后，shutdown線程池，調(diào)用線程池的awaitTermination方法去阻塞等待線程所有執(zhí)行完畢。這個(gè)方法結(jié)束后，自然所有A任務(wù)就執(zhí)行完了，銷毀線程池后執(zhí)行任務(wù)B,再創(chuàng)建新的線程池來執(zhí)行任務(wù)B，以此類推，這樣既用了多線程執(zhí)行速度的優(yōu)點(diǎn)又保證了任務(wù)的執(zhí)行順序。當(dāng)然分布式用隊(duì)列和消費(fèi)者來做的話，就更簡單，只需要監(jiān)控隊(duì)列的已消費(fèi)和待發(fā)送以及消費(fèi)完成數(shù)量，就知道任務(wù)是否完全結(jié)束。以下是多線程上下文任務(wù)的一個(gè)實(shí)現(xiàn)代碼

//創(chuàng)建線程池A,這里線程池改造過可以指定名字
NamedThreadPoolExecutor APool =
                new NamedThreadPoolExecutor(
                        32,
                        128,
                        60,
                        TimeUnit.SECONDS,
                        new LinkedBlockingQueue<>(5),
                        new ThreadPoolExecutor.CallerRunsPolicy(),
                        this.getClass().getSimpleName());
-------------------------------------------------------------
//這里拿到任務(wù)后，把任務(wù)都丟到線程池里去執(zhí)行
List<Object> taskA = getTask();
for(Object task:taskA){
    APool.execute(new CommonRunable(task));
}
//這里調(diào)用shutdown和awaitTermination，并設(shè)置最大超時(shí)，在方法執(zhí)行完后就表示任務(wù)A已經(jīng)完全執(zhí)行完了
APool.shutdown();
APool.awaitTermination(Long.MAX_VALUE,TimeUnit.NANOSECONDS);
//然后這里A已經(jīng)全部執(zhí)行完了，開始執(zhí)行B
NamedThreadPoolExecutor BPool =
                new NamedThreadPoolExecutor(
                        32,
                        128,
                        60,
                        TimeUnit.SECONDS,
                        new LinkedBlockingQueue<>(5),
                        new ThreadPoolExecutor.CallerRunsPolicy(),
                        this.getClass().getSimpleName());
List<Object> taskB = getTask();
for(Object task:taskB){
    BPool.execute(new CommonRunable(task));
}
//這里調(diào)用shutdown和awaitTermination，并設(shè)置最大超時(shí)，在方法執(zhí)行完后就表示任務(wù)B已經(jīng)完全執(zhí)行完了,繼續(xù)執(zhí)行后面的任務(wù)
BPool.shutdown();
BPool.awaitTermination(Long.MAX_VALUE,TimeUnit.NANOSECONDS);


//繼續(xù)執(zhí)行C任務(wù)
...

可暫停并重新開始的定時(shí)任務(wù)

實(shí)現(xiàn)一個(gè)可暫停并重新開始的定時(shí)任務(wù)，首先得考慮場景，如果這個(gè)暫停是主動(dòng)暫停，那需要把這個(gè)任務(wù)執(zhí)行的進(jìn)度和一些中間數(shù)據(jù)存入中間件或者內(nèi)存，然后再次開始的時(shí)候讀取這些中間數(shù)據(jù)就行了。如果這個(gè)暫停的場景包含被動(dòng)暫停，比如任務(wù)中斷，或者進(jìn)程掛了，那就需要在任務(wù)一開始就把中間數(shù)據(jù)和執(zhí)行進(jìn)度都使用中間件存儲(chǔ)，再次開始的時(shí)候會(huì)直接從中間件讀取任務(wù)。一般來說，這里的中間件會(huì)選擇redis，像java中一些list或者set數(shù)據(jù)都是很方便存儲(chǔ)和讀取的。

到此這篇關(guān)于Java中定時(shí)任務(wù)的全方位場景實(shí)現(xiàn)思路分析的文章就介紹到這了,更多相關(guān)Java定時(shí)任務(wù)內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論