【发布时间】：2020-09-10 06:14:52
【问题描述】：
Java中如何使用线程池处理背压？
如何拒绝新任务，使提交的任务不超过 N 个。 N - 提交队列中允许的最大任务数，包括新的、正在运行的、暂停（未完成）的任务。
用例
用户提交运行一段时间的计算任务。有时，有很多用户同时提交任务。如果已提交 N 个任务，如何拒绝新任务。
换句话说，提交的总数（未完成、已启动或未启动）任务不能大于 N。
示例代码
Here is full version 和下面是简短的 sn-ps。
一项长期运行的任务。 计算任务。
public class CalculationTask {
public CalculationTask(final String name) {
this.name = name;
}
public CalculationResult calculate() {
final long waitTimeMs = MIN_WAIT_TIME_MS + RANDOM.nextInt(MAX_WAIT_TIME_MS);
sleep(waitTimeMs);
final int result = Math.abs(RANDOM.nextInt());
final String text = "This is result: " + result;
final CalculationResult calculationResult = new CalculationResult(name, text, result);
System.out.println("Calculation finished: " + calculationResult);
return calculationResult;
}
}
它的结果。 计算结果。
public class CalculationResult {
private final String taskName;
private final String text;
private final Integer number;
// Getters, setters, constructor, toString.
}
这就是我提交作业的方式。 CalculationBroker。
public class CalculationBroker {
private static final int MAX_WORKERS_NUMBER = 5;
private final ExecutorService executorService = Executors.newFixedThreadPool(MAX_WORKERS_NUMBER);
private final Map<String, CalculationResult> calculationCache = new ConcurrentHashMap<>();
public CompletableFuture<CalculationResult> submit(final CalculationTask calculationTask) {
final CalculationResult calculationResultCached = calculationCache.get(calculationTask.getName());
if (calculationResultCached != null) {
return CompletableFuture.completedFuture(calculationResultCached);
}
System.out.println("Calculation submitted: " + calculationTask.getName());
final CompletableFuture<CalculationResult> calculated = CompletableFuture
.supplyAsync(calculationTask::calculate, executorService);
calculated.thenAccept(this::updateCache);
return calculated;
}
private void updateCache(final CalculationResult calculationResult) {
calculationCache.put(calculationResult.getTaskName(), calculationResult);
}
}
这就是我将它们一起运行的方式。 主要。
public class Main {
public static void main(String[] args) {
final int N_TASKS = 100;
final CalculationBroker calculationBroker = new CalculationBroker();
final List<CompletableFuture<CalculationResult>> completableFutures = new ArrayList<>();
for (int i = 0; i < N_TASKS; i++) {
final CalculationTask calculationTask = createCalculationTask(i);
final CompletableFuture<CalculationResult> calculationResultCompletableFuture =
calculationBroker.submit(calculationTask);
completableFutures.add(calculationResultCompletableFuture);
}
calculationBroker.close();
}
private static CalculationTask createCalculationTask(final int counter) {
return new CalculationTask("CalculationTask_" + counter);
}
}
这是输出。
2020-05-23 14:14:53 [main] INFO c.y.t.backperssure.CalculationBroker – Calculation submitted: CalculationTask_97.
2020-05-23 14:14:53 [main] INFO c.y.t.backperssure.CalculationBroker – Calculation submitted: CalculationTask_98.
2020-05-23 14:14:53 [main] INFO c.y.t.backperssure.CalculationBroker – Calculation submitted: CalculationTask_99.
2020-05-23 14:14:54 [pool-1-thread-3] INFO c.y.t.backperssure.CalculationTask – Calculation finished: CalculationResult{taskName='CalculationTask_2', text='This is result: 1081871544', number=1081871544, durationMs=1066}
2020-05-23 14:14:55 [pool-1-thread-1] INFO c.y.t.backperssure.CalculationTask – Calculation finished: CalculationResult{taskName='CalculationTask_0', text='This is result: 1942553785', number=1942553785, durationMs=1885}
2020-05-23 14:14:56 [pool-1-thread-5] INFO c.y.t.backperssure.CalculationTask – Calculation finished: CalculationResult{taskName='CalculationTask_4', text='This is result: 104326011', number=104326011, durationMs=2120}
20
我的发现。
上面的代码等价于 Executors.newFixedThreadPool(n)，但是我们使用固定容量为 100 的 ArrayBlockingQueue，而不是默认的无限 LinkedBlockingQueue。这意味着如果 100 个任务已经排队（并且 n 正在执行），新任务将被被 RejectedExecutionException 拒绝。
ThreadPoolExecutor 使用LinkedBlockingQueue，默认无限制。
正如上面的帖子所暗示的：
final BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(100);
executorService = new ThreadPoolExecutor(n, n, 0L, TimeUnit.MILLISECONDS, queue);
【问题讨论】：
-
只是为了让事情清楚..您不希望能够提交或者您希望您的线程避免开始处理新提交的计算，直到您有空闲线程？？ N.B：两者都可以实现，只是为了能够提供正确的解决方案。
-
好点，谢谢，我更新了我的问题。我想拒绝新任务，所以提交的任务总数（在后台队列中）小于或等于 N。
-
使用背压，使用反应流协议。
-
很高兴看到一个例子，@AlexeiKaigorodov
标签： java concurrency executorservice java.util.concurrent backpressure