Java : Exchanger (同期) - API使用例
2023年04月27日
Exchanger (Java SE 19 & JDK 19) の使用例まとめです。
だいたいのメソッドを網羅済みです。
API仕様のおともにどうぞ。
概要
スレッドをペアにして、ペア内の要素を交換できる同期ポイント。 各スレッドは、exchangeメソッドに入ると何らかのオブジェクトを提供し、パートナ・スレッドと照合して、復帰時にそのパートナのオブジェクトを受け取ります。
Exchanger クラスを使うと、２つのスレッド(ペア) で安全(スレッドセーフ) に要素を交換することができます。
例えば、ペアとして A スレッドと B スレッドがあるとします。
- A スレッドが 要素x を渡すために exchange メソッドを呼び出すと、相手(B)の準備はまだなのでメソッドはそのまま待機します。
- B スレッドが 要素y を渡すために exchange メソッドを呼び出すと、相手(A)の準備はできているので、A, B のメソッドは復帰します。
- A スレッドはメソッドの戻り値として、要素y を受け取ります。
- B スレッドはメソッドの戻り値として、要素x を受け取ります。
// 基準となる時刻
final var current = System.nanoTime();
// 基準となる時刻からの差分を秒として取得
final DoubleSupplier elapsedTime = () -> (System.nanoTime() - current) / 1000000000.0;
final var exchanger = new Exchanger<List<String>>();
try (final var executor = Executors.newFixedThreadPool(2)) {
executor.submit(() -> {
try {
final var x = List.of("a", "b", "c");
System.out.printf("A : task start : x = %s (%f sec.)%n", x, elapsedTime.getAsDouble());
final var ret = exchanger.exchange(x);
System.out.printf("A : exchange OK! : ret = %s (%f sec.)%n",
ret, elapsedTime.getAsDouble());
} catch (InterruptedException e) {
System.out.println("InterruptedException!");
}
});
TimeUnit.SECONDS.sleep(2);
executor.submit(() -> {
try {
final var x = List.of("X", "Y", "Z");
System.out.printf("B : task start : x = %s (%f sec.)%n", x, elapsedTime.getAsDouble());
final var ret = exchanger.exchange(x);
System.out.printf("B : exchange OK! : ret = %s (%f sec.)%n",
ret, elapsedTime.getAsDouble());
} catch (InterruptedException e) {
System.out.println("InterruptedException!");
}
});
}
// 結果
// ↓
//A : task start : x = [a, b, c] (0.002765 sec.)
//B : task start : x = [X, Y, Z] (2.019055 sec.)
//B : exchange OK! : ret = [a, b, c] (2.019486 sec.)
//A : exchange OK! : ret = [X, Y, Z] (2.019660 sec.)
関連記事 :
コンストラクタ
Exchanger ()
新しいExchangerを作成します。
このメソッドの使用例は、exchange(V x) にまとめて記載しました。
そちらのAPI使用例をご参照ください。
メソッド
V exchange (V x)
現在のスレッドは、割り込みが発生しないかぎり、もう一方のスレッドがこの交換ポイントに達するまで待機し、指定されたオブジェクトをそのスレッドに転送して、代わりにもう一方のスレッドのオブジェクトを受け取ります。
// 基準となる時刻
final var current = System.nanoTime();
// 基準となる時刻からの差分を秒として取得
final DoubleSupplier elapsedTime = () -> (System.nanoTime() - current) / 1000000000.0;
final var exchanger = new Exchanger<List<String>>();
try (final var executor = Executors.newFixedThreadPool(2)) {
executor.submit(() -> {
try {
final var x = List.of("a", "b", "c");
System.out.printf("A : task start : x = %s (%f sec.)%n", x, elapsedTime.getAsDouble());
final var ret = exchanger.exchange(x);
System.out.printf("A : exchange OK! : ret = %s (%f sec.)%n",
ret, elapsedTime.getAsDouble());
} catch (InterruptedException e) {
System.out.println("InterruptedException!");
}
});
TimeUnit.SECONDS.sleep(2);
executor.submit(() -> {
try {
final var x = List.of("X", "Y", "Z");
System.out.printf("B : task start : x = %s (%f sec.)%n", x, elapsedTime.getAsDouble());
final var ret = exchanger.exchange(x);
System.out.printf("B : exchange OK! : ret = %s (%f sec.)%n",
ret, elapsedTime.getAsDouble());
} catch (InterruptedException e) {
System.out.println("InterruptedException!");
}
});
}
// 結果
// ↓
//A : task start : x = [a, b, c] (0.002765 sec.)
//B : task start : x = [X, Y, Z] (2.019055 sec.)
//B : exchange OK! : ret = [a, b, c] (2.019486 sec.)
//A : exchange OK! : ret = [X, Y, Z] (2.019660 sec.)
final var exchanger = new Exchanger<List<String>>();
try (final var executor = Executors.newFixedThreadPool(2)) {
final var future = executor.submit(() -> {
try {
final var x = List.of("a", "b", "c");
System.out.println("A : task start : x = " + x);
final var ret = exchanger.exchange(x);
System.out.println("A : exchange OK! : ret = " + ret);
} catch (InterruptedException e) {
System.out.println("InterruptedException!");
}
});
TimeUnit.SECONDS.sleep(2);
System.out.println("future : cancel!");
final var ret = future.cancel(true);
System.out.println("cancelled = " + ret);
}
// 結果
// ↓
//A : task start : x = [a, b, c]
//future : cancel!
//InterruptedException!
//cancelled = true
V exchange (V x, long timeout, TimeUnit unit)
現在のスレッドは、割り込みが発生するか、指定された待機時間が経過しないかぎり、もう一方のスレッドがこの交換ポイントに達するまで待機し、指定されたオブジェクトをそのスレッドに転送して、代わりにもう一方のスレッドのオブジェクトを受け取ります。
timeout と unit パラメータ以外については exchange(V x) の使用例をご参照ください。
// 基準となる時刻
final var current = System.nanoTime();
// 基準となる時刻からの差分を秒として取得
final DoubleSupplier elapsedTime = () -> (System.nanoTime() - current) / 1000000000.0;
final var exchanger = new Exchanger<List<String>>();
try (final var executor = Executors.newFixedThreadPool(2)) {
executor.submit(() -> {
try {
final var x = List.of("a", "b", "c");
System.out.printf("A : task start : x = %s (%f sec.)%n", x, elapsedTime.getAsDouble());
final var ret = exchanger.exchange(x, 5, TimeUnit.SECONDS);
System.out.printf("A : exchange OK! : ret = %s (%f sec.)%n",
ret, elapsedTime.getAsDouble());
} catch (InterruptedException | TimeoutException e) {
System.out.printf("A : %s (%f sec.)%n",
e.getClass().getSimpleName(), elapsedTime.getAsDouble());
}
});
}
// 結果
// ↓
//A : task start : x = [a, b, c] (0.002542 sec.)
//A : TimeoutException (5.018461 sec.)
関連記事
- API 使用例
- BlockingQueue (ブロッキング・キュー)
- Callable
- CancellationException
- ConcurrentHashMap.KeySetView (並列処理用セット)
- ConcurrentLinkedDeque (並列処理用・両端キュー)
- ConcurrentLinkedQueue (並列処理用キュー)
- ConcurrentMap (並列処理用マップ)
- ConcurrentModificationException (並列処理例外)
- ConcurrentSkipListSet (並列処理用セット)
- Condition (同期)
- CopyOnWriteArrayList (並列処理用リスト)
- CopyOnWriteArraySet (並列処理用セット)
- CountDownLatch (同期)
- CyclicBarrier (同期)
- Executor
- ExecutorService
- Executors
- Future
- Future.State
- FutureTask
- InterruptedException (割込み例外)
- Lock (同期)
- Object (オブジェクト)
- Runnable
- Semaphore (セマフォ)
- Thread (スレッド)
- ThreadGroup
- ThreadLocal
- TimeUnit