快捷導(dǎo)航

SpringBoot中使用異步調(diào)度程序的高級(jí)方法

更新時(shí)間：2024年07月04日 09:45:24 作者：Eddie_920

本文主要介紹了SpringBoot中使用異步調(diào)度程序的高級(jí)方法,文中通過(guò)示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧

案例：

有兩個(gè)調(diào)度器，一個(gè)調(diào)度器每3分鐘運(yùn)行一次從數(shù)據(jù)庫(kù)中獲取數(shù)據(jù)，另一個(gè)調(diào)度器每分鐘運(yùn)行一次調(diào)用REST API。

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableAsync;

@SpringBootApplication
public class YourSpringBootApplication {

    public static void main(String[] args) {
        SpringApplication.run(YourSpringBootApplication.class, args);
    }
}

import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

@Component
public class FirstScheduler {

    @Scheduled(fixedRate = 3 * 60 * 1000) // 每3分鐘運(yùn)行一次
    public void getDataFromDB() {
        // 從數(shù)據(jù)庫(kù)獲取數(shù)據(jù)的代碼
    }
}

import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

@Component
public class SecondScheduler {

    @Scheduled(fixedRate = 1 * 60 * 1000) // 每1分鐘運(yùn)行一次
    public void consumeRestApi() {
        // 調(diào)用REST API的代碼
    }
}

問(wèn)題：

當(dāng)?shù)谝粋€(gè)調(diào)度器運(yùn)行時(shí)間超過(guò)預(yù)期時(shí)間，例如執(zhí)行了10分鐘，那么第二個(gè)調(diào)度器則不會(huì)運(yùn)行，一直到第一個(gè)調(diào)度器運(yùn)行結(jié)束后。

上述代碼存在的問(wèn)題，當(dāng)一個(gè)調(diào)度器運(yùn)行時(shí)，第二個(gè)調(diào)度器會(huì)被阻塞。

解決方案：

為了避免第二個(gè)調(diào)度器因第一個(gè)調(diào)度器而被阻塞，可以使用Spring的異步執(zhí)行。這樣做每個(gè)調(diào)度器都將在自己?jiǎn)为?dú)線程中運(yùn)行從而實(shí)現(xiàn)獨(dú)立工作。

實(shí)現(xiàn)代碼：

在使用Spring的異步執(zhí)行支持時(shí)，最佳做法是配置一個(gè)自定義線程池來(lái)控制異步任務(wù)執(zhí)行所用的線程數(shù)量。默認(rèn)情況下Spring使用SimpleAsyncTaskExecutor（這在生產(chǎn)環(huán)境中可能不太合適，因?yàn)樗惶峁└鄬?duì)線程池的控制方法。）
為了解決這個(gè)問(wèn)題建議在應(yīng)用程序中創(chuàng)建一個(gè)自定義線程池bean。以下是實(shí)現(xiàn)代碼：

步驟1：定義一個(gè)配置類(lèi)創(chuàng)建自定義線程池bean。

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

@Configuration
@EnableAsync
public class AsyncConfiguration {

    @Bean(name = "asyncTaskExecutor")
    public ThreadPoolTaskExecutor asyncTaskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(5); // 設(shè)置線程池中的初始線程數(shù)
        executor.setMaxPoolSize(10); // 設(shè)置線程池中的最大線程數(shù)
        executor.setQueueCapacity(25); // 設(shè)置用于保存掛起任務(wù)的隊(duì)列容量
        executor.setThreadNamePrefix("AsyncTask-"); // 設(shè)置線程名前綴
        executor.initialize();
        return executor;
    }
}

步驟2：修改第一個(gè)調(diào)度器，使其使用此自定義線程池。

import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

@Component
public class FirstScheduler {

    @Async("asyncTaskExecutor") // 指定自定義線程池bean名稱(chēng)
    @Scheduled(fixedRate = 3 * 60 * 1000) // 每3分鐘運(yùn)行一次
    public void getDataFromDB() {
        // 從數(shù)據(jù)庫(kù)獲取數(shù)據(jù)的代碼
        // 此方法將在自定義線程池中異步運(yùn)行
    }
}

通過(guò)這種配置，第一個(gè)調(diào)度器方法（getDataFromDB）將在自定義線程池中異步運(yùn)行，而第二個(gè)調(diào)度器方法（consumeRestApi）則在默認(rèn)調(diào)度器的線程中運(yùn)行。

根據(jù)應(yīng)用程序需求和可用系統(tǒng)資源調(diào)整corePoolSize、maxPoolSize和queueCapacity值。線程池配置對(duì)應(yīng)用程序的性能產(chǎn)生比較重要的影響，因此需要適當(dāng)調(diào)整這些值。

要使第二個(gè)調(diào)度器也使用自定義線程池進(jìn)行異步執(zhí)行，需要在consumeRestApi方法的@Async注解中添加taskExecutor屬性。這樣確保兩個(gè)調(diào)度器都在相同的自定義線程池中異步運(yùn)行。以下是更新后的代碼：

import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

@Component
public class SecondScheduler {

    @Async("asyncTaskExecutor") // 指定自定義線程池bean名稱(chēng)
    @Scheduled(fixedRate = 1 * 60 * 1000) // 每1分鐘運(yùn)行一次
    public void consumeRestApi() {
        // 調(diào)用REST API的代碼
        // 此方法將在自定義線程池中異步運(yùn)行
    }
}

通過(guò)這種設(shè)置，第一個(gè)調(diào)度器（getDataFromDB）和第二個(gè)調(diào)度器（consumeRestApi）都將在相同的自定義線程池中異步運(yùn)行。這將允許它們獨(dú)立地工作，即使其中一個(gè)任務(wù)需要更長(zhǎng)時(shí)間來(lái)完成。

使用自定義線程池
記錄錯(cuò)誤消息，當(dāng)所需線程池大小 > 配置的線程池大小時(shí)：
要在所需線程池大小超過(guò)配置的線程池大小時(shí)記錄錯(cuò)誤消息，可以利用Spring的ThreadPoolTaskExecutor的RejectedExecutionHandler。當(dāng)線程池的任務(wù)隊(duì)列已滿(mǎn)且線程池?zé)o法接受更多任務(wù)時(shí)，將調(diào)用此處理程序?？梢允褂么嘶卣{(diào)來(lái)記錄錯(cuò)誤消息。

以下是更新的配置類(lèi)，其中包含RejectedExecutionHandler：

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.ThreadPoolExecutor;

@Configuration
@EnableAsync
@Sl4J
public class AsyncConfiguration {

    @Bean(name = "asyncTaskExecutor")
    public ThreadPoolTaskExecutor asyncTaskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(5); // 設(shè)置線程池中的初始線程數(shù)
        executor.setMaxPoolSize(10); // 設(shè)置線程池中的最大線程數(shù)
        executor.setQueueCapacity(25); // 設(shè)置用于保存掛起任務(wù)的隊(duì)列容量
        executor.setThreadNamePrefix("AsyncTask-"); // 設(shè)置線程名前綴

        // 設(shè)置RejectedExecutionHandler以記錄錯(cuò)誤消息
        executor.setRejectedExecutionHandler((Runnable r, ThreadPoolExecutor e) -> {
            // 記錄錯(cuò)誤消息，隊(duì)列已滿(mǎn)，任務(wù)被拒絕
            // 根據(jù)需要自定義此消息
            log.error("任務(wù)被拒絕：線程池已滿(mǎn)。請(qǐng)?jiān)黾泳€程池大小。");
        });

        executor.initialize();
        return executor;
    }
}

通過(guò)這種配置，當(dāng)線程池的隊(duì)列已滿(mǎn)且嘗試提交額外任務(wù)時(shí)，RejectedExecutionHandler會(huì)被觸發(fā)。
你可以根據(jù)應(yīng)用程序的需求自定義錯(cuò)誤消息或采取其他操作。

請(qǐng)注意，設(shè)置正確的線程池大小和隊(duì)列容量對(duì)于應(yīng)用程序的性能和資源利用率至關(guān)重要。

如果因?yàn)殛?duì)列容量不足而任務(wù)持續(xù)被拒絕則需要增加線程池大小或調(diào)整隊(duì)列容量。

總結(jié)：

通過(guò)配置共享的自定義線程池、利用@Async注解以及整合RejectedExecutionHandler，可以使應(yīng)用程序有效地管理和執(zhí)行多個(gè)調(diào)度器并發(fā)從而確保調(diào)度器獨(dú)立運(yùn)行，并且系統(tǒng)能夠優(yōu)雅地響應(yīng)線程池限制。

到此這篇關(guān)于SpringBoot中使用異步調(diào)度程序的高級(jí)方法的文章就介紹到這了,更多相關(guān)SpringBoot 異步調(diào)度內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: