私はフォーク/ジョインプールの1つが動作しなくなった現在、生産に2回現れているバグを持っています。fork-joinがうまくいっているときにワーカースレッドはありませんか?
これまでのところ、タスクのキューがいっぱいになってタスク結果の流れが止まっている理由を説明するための結論です。私は、タスクプロデューサスレッドがフォーク/ジョインのサブミットを完了するのを待っているスレッドダンプを持っていますが、何もしないForkJoinPoolワーカースレッドはありません。
"calc-scheduling-pool-4-thread-2" #65 prio=5 os_prio=0 tid=0x00000000102e39f0 nid=0x794a in Object.wait() [0x00002ad900a06000]
java.lang.Thread.State: WAITING (on object monitor)
at java.lang.Object.wait(Native Method)
at java.util.concurrent.ForkJoinTask.externalAwaitDone(ForkJoinTask.java:334)
- locked <0x000000061ad08708> (a com.....Engine$Calculation)
at java.util.concurrent.ForkJoinTask.doJoin(ForkJoinTask.java:391)
at java.util.concurrent.ForkJoinTask.join(ForkJoinTask.java:719)
at java.util.concurrent.ForkJoinPool.invoke(ForkJoinPool.java:2613)
at com...Engine.calculateSinceLastBatch(Engine.java:141)
私がやっていることに関係なく、これは正しくないはずですか?スレッドダンプは、初期状態が検出されてから数時間後です。私はランタイムで2つの他のForkJoinPoolsが両方とも通常多くのワーカースレッドが存在する状態で動作しています。
このプールの並列性は1です(私はそれが愚かだと知っていますが、フォーク/ジョインプールの正確性を損なうべきではありません)。タスクキューがいっぱいになり、スレッドダンプがワーカーを検出しなくなるまで、エラーや例外は検出されません。
他にもこれを見ましたか?いずれか私は何かが不足しているか、フォーク/参加のバグがありません。
ランタイムは、コード
でJava 8
アップデートでこれが私たちが制作に参加/フォークを使用しているかの合理的な単純化です。私たちは、これが私に語っ待ち時間はjava.util.concurrent.ForkJoinTask.externalAwaitDone(ForkJoinTask.java:334)である1
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.*;
public class Engine {
BlockingQueue<Calculation> externalQueue = new LinkedBlockingQueue<>(100000);
ScheduledExecutorService scheduling = Executors.newScheduledThreadPool(3);
static ForkJoinPool forkJoin = new ForkJoinPool(1);
public static void main(String[] args) {
new Engine().start();
}
void start() {
final AtomicInteger batch = new AtomicInteger(0);
// data comes in from external systems
scheduling.scheduleWithFixedDelay(
() -> produceData(batch.getAndIncrement()),
500,
500,
TimeUnit.MILLISECONDS);
// internal scheduling processes data with a fixed delay
scheduling.scheduleWithFixedDelay(
this::calculate,
1000,
1000,
TimeUnit.MILLISECONDS);
}
void produceData(final int batch) {
System.out.println(Thread.currentThread().getName() + " => submitting data for batch " + batch);
Stream<Integer> data = IntStream.range(0, 10).boxed();
data.map((i) -> new Calculation(batch, i)).forEach(externalQueue::offer);
}
void calculate() {
int available = externalQueue.size();
List<Calculation> tasks = new ArrayList<>(available);
externalQueue.drainTo(tasks);
// invoke will block for the results to be calculated before continuing
forkJoin.invoke(new CalculationTask(tasks, 0, tasks.size()));
System.out.println("done with calculations at " + new Date());
}
static class CalculationTask extends RecursiveAction {
static int MIN_CALCULATION_THRESHOLD = 3;
List<Calculation> tasks;
int start;
int end;
CalculationTask(List<Calculation> tasks, int start, int end) {
this.tasks = tasks;
this.start = start;
this.end = end;
}
// if below a threshold, calculate here, else fork to new CalculationTasks
@Override
protected void compute() {
int work = end - start;
if (work <= threshold()) {
for (int i = start; i < end; i++) {
Calculation calc = tasks.get(i);
calc.calculate();
}
return;
}
invokeNewActions();
}
int threshold() {
return Math.max(tasks.size()/forkJoin.getParallelism()/2, MIN_CALCULATION_THRESHOLD);
}
void invokeNewActions() {
invokeAll(
new CalculationTask(tasks, start, middle()),
new CalculationTask(tasks, middle(), end));
}
int middle() {
return (start + end)/2;
}
}
static class Calculation {
int batch;
int data;
Calculation(int batch, int data) {
this.batch = batch;
this.data = data;
}
void calculate() {
// does some work and pushes results to a listener
System.out.println(Thread.currentThread().getName() + " => calculation complete on batch " + batch
+ " for " + data);
}
}
}
キューとは何ですか?どのリリース、Java7または8?少しのコードも役立つかもしれません。 – edharned
ご迷惑をおかけして申し訳ございません。私は大量のタスクをキューに入れ、間隔をおいて排水され、フォーク/ジョインに割り振られて分解され、実行されます。 プロダクション環境はJava 8です。私はいくつかのコードを提供しようとすることができますが、フォーク/ジョインの使い方がチュートリアルのように見えます。 –
並列処理を1より大きい値に設定しようとしましたか?私はこれがあなたが探している答えではないことを知っていますが、再現するテストがなければ、誰かが確かな答えを持つことはまずありません。あなたの前提は正しいものであり、過去に奇妙なことが報告されています。 –