我們有一個(gè) Spring 應(yīng)用程序，我想添加一個(gè)服務(wù)，該服務(wù)將使用多個(gè)線程處理 10Ks ID，但將作為后臺(tái)進(jìn)程而不影響生產(chǎn)實(shí)時(shí)性。服務(wù)將更新數(shù)據(jù)庫(kù)并發(fā)送外部提供商請(qǐng)求。我不希望服務(wù)影響生產(chǎn)性能/計(jì)時(shí)，我想以低優(yōu)先級(jí)對(duì)每個(gè) ID 執(zhí)行操作我希望對(duì)超出此特定執(zhí)行器范圍的所有其他線程設(shè)置低優(yōu)先級(jí)。答案使用ThreadPoolExecutor與我的案例更相關(guān)嗎？ThreadPoolExecutor threadPool = new ThreadPoolExecutor(1, numOfWorkerThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());threadPool.setThreadFactory(new OpJobThreadFactory(Thread.NORM_PRIORITY-2));public final static class OpJobThreadFactory implements ThreadFactory {?private int priority;?public OpJobThreadFactory(int priority) {? this(priority, true);}@Overridepublic Thread newThread(Runnable r) {? Thread t = new Thread(r, namePrefix + threadNumber.getAndIncrement());? t.setDaemon(daemon);? t.setPriority(priority);?}}也許甚至使用Thread.MIN_PRIORITY或者我可以使用Executors.newCachedThreadPool()創(chuàng)建一個(gè)線程池，該線程池根據(jù)需要?jiǎng)?chuàng)建新線程，但會(huì)重用以前構(gòu)造的線程（當(dāng)它們可用時(shí)）。這些池通常會(huì)提高執(zhí)行許多短期異步任務(wù)的程序的性能。我還應(yīng)該使用Spring bean嗎？因?yàn)槲倚枰葱?請(qǐng)求創(chuàng)建池，所以似乎不需要/錯(cuò)誤編輯?我應(yīng)該使用Spring Actuator來獲取此任務(wù)還是其他監(jiān)控工具？Spring Boot Actuator 模塊通過提供健康檢查、審計(jì)、指標(biāo)收集、HTTP 跟蹤等生產(chǎn)就緒功能來幫助您監(jiān)控和管理 Spring Boot 應(yīng)用程序。所有這些功能都可以通過 JMX 或 HTTP 端點(diǎn)訪問。
2 回答
達(dá)令說
TA貢獻(xiàn)1821條經(jīng)驗(yàn) 獲得超6個(gè)贊
我想澄清一下這個(gè)問題
什么是線程優(yōu)先級(jí)？根據(jù) java SE 文檔
每個(gè)線程都有一個(gè)優(yōu)先級(jí)。具有較高優(yōu)先級(jí)的線程優(yōu)先于具有較低優(yōu)先級(jí)的線程執(zhí)行。
即使您創(chuàng)建具有優(yōu)先級(jí)的線程，它也不能完全保證優(yōu)先級(jí)較低的線程首先執(zhí)行，您可能必須阻塞優(yōu)先級(jí)較低的線程，直到執(zhí)行其他線程
對(duì)于小型 java 程序，您可以自己處理線程執(zhí)行，但對(duì)于較大的程序，建議您使用 vanilla Java 中的 Executor Framework（來自 java.util.concurrent 包）或使用 spring TaskExecutor。通過使用這兩個(gè)框架，您可以異步執(zhí)行任務(wù)，即作為主任務(wù)的一部分在后臺(tái)執(zhí)行它們。
對(duì)生產(chǎn)的影響：
例如，主要任務(wù)將是調(diào)用您的其余端點(diǎn)（即/account），并在調(diào)用帳戶端點(diǎn)時(shí)您想要向客戶發(fā)送歡迎電子郵件，這是第三方 API 調(diào)用，可以使用 Executor Framework 或 Spring 異步執(zhí)行TaskExecutor 異步執(zhí)行它們，即作為后臺(tái)進(jìn)程，它們不會(huì)對(duì)當(dāng)前 API 產(chǎn)生影響，但肯定會(huì)對(duì)生產(chǎn)服務(wù)器產(chǎn)生影響，因?yàn)槟谕粋€(gè) JVM 中運(yùn)行線程并且它們共享公共內(nèi)存。如果創(chuàng)建了許多線程但沒有銷毀，那么服務(wù)器肯定會(huì)崩潰。
因此，使用 Executor Framework 或 Spring TaskExecutor 并不能保證它不會(huì)影響您當(dāng)前的生產(chǎn)，它肯定會(huì)提高所調(diào)用的其余 API 的性能。因?yàn)樗钱惒綀?zhí)行的并且關(guān)于您的其他問題
我可以使用 Executors.newCachedThreadPool()
是的，如果您有許多短期任務(wù)，例如更新數(shù)據(jù)庫(kù)中的單個(gè)列或僅觸發(fā)一次休息端點(diǎn)，并且它不適合批量加載或執(zhí)行某些更新 10000 條記錄的后端作業(yè)，因?yàn)樗鼤?huì)創(chuàng)建更大的數(shù)量每個(gè)任務(wù)都有多個(gè)線程，您肯定會(huì)遇到內(nèi)存問題。
慕萊塢森
TA貢獻(xiàn)1810條經(jīng)驗(yàn) 獲得超4個(gè)贊
這是一種創(chuàng)建可配置任務(wù)“堆”并始終將主任務(wù)保留在堆頂部的方法。
總結(jié)這個(gè)過程，您應(yīng)該創(chuàng)建一個(gè)自定義任務(wù)執(zhí)行器。首先，您需要?jiǎng)?chuàng)建一個(gè) ThreadPoolTaskExecutor?bean，并覆蓋一個(gè)方法。需要修改的屬性有：CorePoolSize（初始線程數(shù)）、QueueCapacity（隊(duì)列中等待的線程數(shù)）和MaxPoolSize（最大線程數(shù)）。使用這些參數(shù)，您可以配置應(yīng)用程序限制，以使該服務(wù)不會(huì)影響生產(chǎn)性能。
?@Bean("CustomTaskExecutor")
? public TaskExecutor threadPoolTaskExecutor(
? ? ? ? ? @Value("${spring.async.core-pool-size}") int corePoolSize,
? ? ? ? ? @Value("${spring.async.max-pool-size}") int maxPoolSize,
? ? ? ? ? @Value("${spring.async.queue-capacity}") int queueCapacity) {
? ? ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor() {
? ? ? @Override
? ? ? protected BlockingQueue<Runnable> createQueue(int queueCapacity) {
? ? ? ? return new PriorityBlockingQueue<Runnable>(queueCapacity);
? ? ? }
? ? };
? ? executor.setCorePoolSize(corePoolSize);
? ? executor.setMaxPoolSize(maxPoolSize);
? ? executor.setQueueCapacity(queueCapacity);
? ? return executor;
? }
之后，您需要制定任務(wù)執(zhí)行者可以理解的優(yōu)先級(jí)。為此，我們需要?jiǎng)?chuàng)建兩個(gè)類： 1) 一個(gè)實(shí)現(xiàn) Runnable 接口的自定義類，該類將運(yùn)行任務(wù) 2) 一個(gè)擴(kuò)展 FutureTask 并實(shí)現(xiàn) Comparable 接口的包裝類，以便任務(wù)執(zhí)行器可以理解任務(wù)的優(yōu)先級(jí)選擇邏輯任務(wù)
public class Task implements Runnable {
? ? private Consumer<Job> jobConsumer;
? ? private Job job;
? ? ? public Job getJob() {
? ? ? ? return this.job;
? ? ? }
? ? ? public Task(Consumer<Job> jobConsumer, Job job) {
? ? ? ? this.jobConsumer = jobConsumer;
? ? ? ? this.job = job;
? ? ? }
? ? ? @Override
? ? ? public void run() {
? ? ? ? this.jobConsumer.accept(job);
? ? ? }
? ? }
然后你就有了 FutureCustomTask 類：
public class FutureCustomTask extends FutureTask<FutureCustomTask> implements Comparable<FutureCustomTask> {
private Task task;
public FutureCustomTask(Task task) {
? ? super(task, null);
? ? this.task = task;
? }
@Override
? public int compareTo(FutureCustomTask o) {
? ? return task.getJob().getPriority().compareTo(o.task.getJob().getPriority());
? }
}
為了執(zhí)行任務(wù)執(zhí)行器需要自動(dòng)裝配。然后，您可以創(chuàng)建 Task 對(duì)象，將其包裝在 FutureCustomTask 中，并將其傳遞給 TaskExecutor。代碼應(yīng)如下所示：
@Autowired
private TaskExecutor taskExecutor;
@Autowired
private JobBusiness jobBusiness;
...
Task task = new Task(jobBusiness::performSomethingOn, job);
taskExecutor.execute(new FutureCustomTask(task));
添加回答
舉報(bào)
0/150
提交
取消