😵 🤳🏼 🌕 シンクロナイザー参照java.util.concurrent。* 🧔🏿 🌻 💸

この出版物の目的は、java.util.concurrentパッケージのシンクロナイザーを完全に分析することではありません。まず第一に、トピックへのエントリを容易にし、スレッドを同期するためのクラスの実用的なアプリケーションの可能性を示す参考書としてそれを書いています（以降、スレッド=スレッド）。

java.util.concurrentには、機能に応じてグループに分類できる多くの異なるクラスがあります。ConcurrentCollections、Executors、Atomicsなどです。これらのグループの1つはシンクロナイザーです。

シンクロナイザーは、フローを同期するための補助ユーティリティです。これにより、開発者は、フローの動作を調整および/または制限し、基本的な言語プリミティブ（モニター）よりも高い抽象度を提供できます。

セマフォ

セマフォシンクロナイザは、セマフォ同期パターンを実装します。共有リソースへのアクセスを制限する必要がある場合、ほとんどの場合、セマフォが必要です。このクラスのコンストラクタ（ Semaphore(int permits)またはSemaphore(int permits, boolean fair) ）は、セマフォが指定されたリソースを同時に使用できるスレッドの数を渡す必要があります。

アクセスはカウンタによって制御されます。最初は、カウンタ値はint permitsです。ストリームが特定のコードブロックに入ると、カウンタ値は1減少し、ストリームがそれを離れると増加します。カウンター値がゼロの場合、誰かがブロックを離れるまで現在のフローがブロックされます（ permits = 1生活の例として、診療所のオフィスにキューを持ち込むことができます：患者がオフィスを離れると、ランプが点滅し、次の患者が入室します））

セマフォの公式ドキュメント。

セマフォの例

次の例を考えてみましょう。一度に5台までしか駐車できない駐車場があります。駐車場が完全に満車の場合、新しく到着した車は、少なくとも1席が空くまで待たなければなりません。その後、彼は駐車することができます。

 import java.util.concurrent.Semaphore; public class Parking { //   - true,  - false private static final boolean[] PARKING_PLACES = new boolean[5]; //  "",     //aquire()       private static final Semaphore SEMAPHORE = new Semaphore(5, true); public static void main(String[] args) throws InterruptedException { for (int i = 1; i <= 7; i++) { new Thread(new Car(i)).start(); Thread.sleep(400); } } public static class Car implements Runnable { private int carNumber; public Car(int carNumber) { this.carNumber = carNumber; } @Override public void run() { System.out.printf(" №%d   .\n", carNumber); try { //acquire()          , //   ,        , //     SEMAPHORE.acquire(); int parkingNumber = -1; //     synchronized (PARKING_PLACES){ for (int i = 0; i < 5; i++) if (!PARKING_PLACES[i]) { //   PARKING_PLACES[i] = true; //  parkingNumber = i; //  ,   System.out.printf(" №%d    %d.\n", carNumber, i); break; } } Thread.sleep(5000); //  ,   synchronized (PARKING_PLACES) { PARKING_PLACES[parkingNumber] = false;//  } //release(), ,   SEMAPHORE.release(); System.out.printf(" №%d  .\n", carNumber); } catch (InterruptedException e) { } } } }

プログラムの結果

1号車は駐車場まで運転しました。
1号車は0に駐車されました。
2号車は駐車場まで運転しました。
2号車は1番に駐車されました。
3号車は駐車場まで運転しました。
3号車は2番に駐車されました。
4号車は駐車場まで運転しました。
4号車は3番に駐車されました。
5号車は駐車場まで運転しました。
5号車は4番に駐車されました。
6号車は駐車場まで運転しました。
7号車は駐車場まで運転しました。
1号車が駐車場を出ました。
車番号6は0に駐車されました。
2号車が駐車場を出ました。
7番の車が所定の場所に駐車しています。
3号車が駐車場を出ました。
4号車が駐車場を出ました。
5号車が駐車場を出ました。
6号車が駐車場を出ました。
7号車が駐車場を出ました。

セマフォはこの問題を解決するのに最適です：駐車スペースがない場合（カウンターは0）、車（ストリーム）が駐車（特定のコードブロックに移動し、共有リソースを使用）することを許可しません。セマフォクラスは複数のキャプチャと解放をサポートすることに注意してください一度に許可されますが、このタスクは必要ありません。

CountDownLatch

CountDownLatch（カウントダウン付きロック）を使用すると、コードブロック内の任意の数のスレッドが、他のスレッドで特定の数の操作が完了するまで待機してから、アクティビティを継続することができます。コンストラクターCountDownLatch（ CountDownLatch(int count) ）は、ロックがブロックされたスレッドを「解放」するために実行する必要がある操作の数を渡す必要があります。

ブロッキングフローは、カウンターを使用して削除されます。アクティブなストリームは、特定の操作を実行すると、カウンターの値を減らします。カウンターが0に達すると、待機中のすべてのスレッドがロック解除され、実行が継続されます（人生のCountDownLatchの例は、エクスカーショングループのコレクションです。特定の人数が集まるまで、エクスカーションは開始されません）。

CountDownLatchの公式ドキュメント。

CountDownLatchの例

次の例を考えてみましょう。カーレースをしたいです。 5台の車がレースに参加します。レースを開始するには、次の条件が満たされている必要があります。

5台の車はそれぞれ、スタートラインまで走りました。
コマンド「To start！」が与えられました。
コマンド「警告！」が与えられました。
コマンド「March！」が与えられました。

すべての車が同時にスタートすることが重要です。

 import java.util.concurrent.CountDownLatch; public class Race { // CountDownLatch  8 "" private static final CountDownLatch START = new CountDownLatch(8); //    private static final int trackLength = 500000; public static void main(String[] args) throws InterruptedException { for (int i = 1; i <= 5; i++) { new Thread(new Car(i, (int) (Math.random() * 100 + 50))).start(); Thread.sleep(1000); } while (START.getCount() > 3) //,     Thread.sleep(100); //  .  ,  100ms Thread.sleep(1000); System.out.println(" !"); START.countDown();// ,    1 Thread.sleep(1000); System.out.println("!"); START.countDown();// ,    1 Thread.sleep(1000); System.out.println("!"); START.countDown();// ,    1 //   ,     //  } public static class Car implements Runnable { private int carNumber; private int carSpeed;//,     public Car(int carNumber, int carSpeed) { this.carNumber = carNumber; this.carSpeed = carSpeed; } @Override public void run() { try { System.out.printf(" №%d    .\n", carNumber); //     -   //   1 START.countDown(); // await()  ,  ,   ,  // CountDownLatch    0 START.await(); Thread.sleep(trackLength / carSpeed);//    System.out.printf(" №%d !\n", carNumber); } catch (InterruptedException e) { } } } }

プログラムの結果

1号車がスタートラインまで走りました。
2号車はスタートラインまで走りました。
3号車はスタートラインまで走りました。
4号車はスタートラインまで走りました。
5号車はスタートラインまで走りました。
はじめに！
注意！
3月！
4号車が完成しました！
1号車が終了しました！
3号車が完成しました！
5号車が完成しました！
2号車が完成しました！

CountDownLatchはさまざまな同期スキームで使用できます。たとえば、1つのスレッドが作業を行っている間、他のスレッドを待機させたり、逆に他のスレッドが作業を行うのを待機させたりできます。

CyclicBarrier

CyclicBarrierは、バリア同期パターンを実装します。ループバックバリアは、指定された数の並列スレッドが出会ってブロックする同期ポイントです。すべての利益が流れるとすぐに、オプションのアクションが実行され（またはバリアがそれなしで初期化された場合は実行されません）、完了後、バリアが解除され、待機中のフローが「解放」されます。バリアコンストラクター（ CyclicBarrier(int parties)およびCyclicBarrier(int parties, Runnable barrierAction) ）は、「会う」べきパーティーの数と、オプションで、パーティーが会うとき、ただしパーティーが会う前に発生するアクションを渡す必要があります。「リリースされました。」

バリアはCountDownLatchに似ていますが、2つの主な違いは、カウンタがゼロになった後は「ロック」を再利用できず、バリアが壊れた後でも再び使用できることです。 CyclicBarrierは、スレッドが実行された後にのみスレッドを「収集」 join()メソッドの代替です。

CyclicBarrierの公式ドキュメント。

CyclicBarrierのケーススタディ

次の例を考えてみましょう。フェリーサービスがあります。フェリーは一度に3台の車を輸送できます。フェリーを再度運転しないようにするには、少なくとも3台の車が交差点に集まるときにフェリーを送る必要があります。

 import java.util.concurrent.CyclicBarrier; public class Ferry { private static final CyclicBarrier BARRIER = new CyclicBarrier(3, new FerryBoat()); //      ,   ,  //    .  ,   . public static void main(String[] args) throws InterruptedException { for (int i = 0; i < 9; i++) { new Thread(new Car(i)).start(); Thread.sleep(400); } } //,        public static class FerryBoat implements Runnable { @Override public void run() { try { Thread.sleep(500); System.out.println("  !"); } catch (InterruptedException e) { } } } //,     public static class Car implements Runnable { private int carNumber; public Car(int carNumber) { this.carNumber = carNumber; } @Override public void run() { try { System.out.printf(" №%d    .\n", carNumber); //        ,    await() //    ,        BARRIER.await(); System.out.printf(" №%d  .\n", carNumber); } catch (Exception e) { } } } }

プログラムの結果

車番号0はフェリーまで運転しました。
1号車がフェリーまで運転しました。
2号車がフェリーまで走りました。
3号車がフェリーまで走りました。
フェリーフェリーカー！
2号車は動き続けました。
1号車は動き続けました。
車番号0は動き続けました。
4号車がフェリーまで走りました。
5号車がフェリーまで走りました。
6号車がフェリーまで運転しました。
フェリーフェリーカー！
5号車は動き続けました。
4号車は動き続けました。
3号車は動き続けました。
7号車がフェリーまで走りました。
8号車がフェリーまで走りました。
フェリーフェリーカー！
8号車は動き続けました。
6号車は動き続けました。
7号車は動き続けました。

3つのストリームがawait()メソッドに到達すると、バリアアクションがトリガーされ、蓄積されたフェリーからの3台の車がフェリーされます。この後、新しいサイクルが始まります。

エクスチェンジャー<V>

両方のストリームの操作のある時点で、2つのストリーム間でデータを交換するには、交換機（交換機）が必要になる場合があります。エクスチェンジャーは一般化されたクラスであり、転送するオブジェクトのタイプによってパラメーター化されます。

エクスチェンジャーは、スレッドのペアの同期ポイントです。 exchange()メソッドを呼び出すスレッドexchange()ブロックされ、別のスレッドを待機します。別のスレッドが同じメソッドを呼び出すと、オブジェクトが交換されます。各オブジェクトは、 exchange()メソッドで他のオブジェクトへの引数を受け取ります。交換機がnull値の転送をサポートしていることは注目に値します。これにより、オブジェクトを一方向に、または単純に2つのストリームの同期ポイントとして転送することができます。

Exchangerの公式ドキュメント。

エクスチェンジャーの例

次の例を考えてみましょう。 2つのトラックがあります：1つはポイントAからポイントDに、もう1つはポイントBからポイントCに行きます。道路ADとBCはポイントEで交差します。ポイントAとBから、荷物はポイントCとDに配達されます。このため、ポイントEのトラック関連するパッケージを満たし、交換する必要があります。

 import java.util.concurrent.Exchanger; public class Delivery { // ,     String private static final Exchanger<String> EXCHANGER = new Exchanger<>(); public static void main(String[] args) throws InterruptedException { String[] p1 = new String[]{"{ A->D}", "{ A->C}"};//   1-  String[] p2 = new String[]{"{ B->C}", "{ B->D}"};//   2-  new Thread(new Truck(1, "A", "D", p1)).start();// 1-     D Thread.sleep(100); new Thread(new Truck(2, "B", "C", p2)).start();// 2-      } public static class Truck implements Runnable { private int number; private String dep; private String dest; private String[] parcels; public Truck(int number, String departure, String destination, String[] parcels) { this.number = number; this.dep = departure; this.dest = destination; this.parcels = parcels; } @Override public void run() { try { System.out.printf("  №%d : %s  %s.\n", number, parcels[0], parcels[1]); System.out.printf(" №%d    %s   %s.\n", number, dep, dest); Thread.sleep(1000 + (long) Math.random() * 5000); System.out.printf(" №%d    .\n", number); parcels[1] = EXCHANGER.exchange(parcels[1]);//  exchange()     //    exchange(),      System.out.printf("  №%d     %s.\n", number, dest); Thread.sleep(1000 + (long) Math.random() * 5000); System.out.printf(" №%d   %s  : %s  %s.\n", number, dest, parcels[0], parcels[1]); } catch (InterruptedException e) { } } } }

プログラムの結果

以下がトラック1番に積載されました：{パッケージA-> D}および{パッケージA-> C}。
トラックNo. 1は、ポイントAからポイントDまで運転しました。
トラックNo. 2に搭載：{パッケージB-> C}および{パッケージB-> D}。
トラック＃2は、ポイントBからポイントCまで運転しました。
トラック1番がポイントEに到着しました。
トラックNo. 2はポイントEに到着しました。
アイテムCのトラックはトラック2に移動しました。
アイテムDのトラックはトラック1に移動しました。
トラックNo. 2がCに到着し、配達されました：{パッケージB-> C}および{パッケージA-> C}。
トラックNo. 1がDに到着し、配達されました：{パッケージA-> D}および{パッケージB-> D}。

ご覧のとおり、1台のトラック（1つのストリーム）がポイントE（同期ポイントに到達）に到着すると、別のトラック（別のストリーム）がポイントE（同期ポイントに到達）に到達するまで待機します。この後、小包交換が行われ（ストリング）、両方のトラック（ストリーム）がパス（作業）を継続します。

フェイザー

CyclicBarrierのようなPhaser（phaser）は、 Barrier同期パターンの実装ですが、CyclicBarrierとは異なり、より柔軟性があります。このクラスを使用すると、共通アクションの単一のフェーズまたはステージを表すスレッドを同期できます。 CyclicBarrierと同様に、Phaserはメンバーストリームが出会う同期ポイントです。すべての関係者が到着すると、Phaserは次のフェーズに進み、再び完了するのを待ちます。

PhaserとCyclicBarrierを比較すると、Phaserの次の重要な機能を強調できます。

各フェーズ（同期サイクル）には番号があります。
参加者の数は厳密に設定されておらず、異なる場合があります。フローは参加者として登録され、その参加を取り消す場合があります。
参加者は、他のすべての参加者がバリアに集まるまで待つ必要はありません。仕事を続けるには、到着について知らせるだけで十分です。
ランダムな目撃者は、バリア内のアクティビティを監視できます。
フローは、それが克服されることを期待するために、バリアの当事者ではない場合があります。
フェイザーにはオプションのアクションはありません。

Phaserオブジェクトは、コンストラクターのいずれかを使用して作成されます。

 Phaser() Phaser(int parties)

Partyパラメーターは、アクションのフェーズを実行するパーティーの数を示します。最初のコンストラクターは、側面のないPhaserオブジェクトを作成し、この場合のバリアも「閉じられます」。 2番目のコンストラクターは、コンストラクターに渡されるサイドの数を登録します。バリアは、すべての関係者が到着したとき、または最後の参加者が削除されたときに開きます。（Phaserクラスには、まだ親Phaserオブジェクトが渡されるコンストラクタがありますが、それらは考慮しません。）

主な方法：

int register（） -フェーズを実行する新しいメンバーを登録します。現在のフェーズ番号を返します。
int getPhase（） -現在のフェーズの番号を返します。
int reachAndAwaitAdvance（） -スレッドがフェーズを完了したことを示します。他のすべての関係者がこのフェーズを完了するまで、フローは中断されます。 CyclicBarrier.await()正確な対応物。現在のフェーズ番号を返します。
int Arrivate（） -パーティがフェーズを完了したことを報告し、フェーズ番号を返します。このメソッドが呼び出されると、スレッドは停止しませんが、実行を継続します。
int ArrivedAndDeregister（） -パーティによるすべてのフェーズの完了を報告し、登録から削除します。現在のフェーズ番号を返します。
int awaitAdvance（int phase）-phaseが現在のフェーズの数と等しい場合、それが発生したスレッドを終了するまで一時停止します。それ以外の場合は、すぐに引数を返します。

Phaserの公式ドキュメント。

フェイザーの例

次の例を考えてみましょう。 5つの停留所があります。最初の4人は乗客を立たせてバスを待つことができます。バスは公園を出て、各停留所でしばらく止まります。最終停車後、バスは公園に行きます。乗客を迎え、適切な停留所で降ろす必要があります。

 import java.util.ArrayList; import java.util.concurrent.Phaser; public class Bus { private static final Phaser PHASER = new Phaser(1);//    // 0  6 -   , 1 - 5  public static void main(String[] args) throws InterruptedException { ArrayList<Passenger> passengers = new ArrayList<>(); for (int i = 1; i < 5; i++) { //    if ((int) (Math.random() * 2) > 0) passengers.add(new Passenger(i, i + 1));//     if ((int) (Math.random() * 2) > 0) passengers.add(new Passenger(i, 5)); //     } for (int i = 0; i < 7; i++) { switch (i) { case 0: System.out.println("   ."); PHASER.arrive();//  0  1  -  break; case 6: System.out.println("   ."); PHASER.arriveAndDeregister();//  ,   break; default: int currentBusStop = PHASER.getPhase(); System.out.println(" № " + currentBusStop); for (Passenger p : passengers) //,      if (p.departure == currentBusStop) { PHASER.register();// ,      p.start(); //   } PHASER.arriveAndAwaitAdvance();//    } } } public static class Passenger extends Thread { private int departure; private int destination; public Passenger(int departure, int destination) { this.departure = departure; this.destination = destination; System.out.println(this + "    № " + this.departure); } @Override public void run() { try { System.out.println(this + "   ."); while (PHASER.getPhase() < destination) //      () PHASER.arriveAndAwaitAdvance(); //        Thread.sleep(1); System.out.println(this + "  ."); PHASER.arriveAndDeregister(); //     } catch (InterruptedException e) { } } @Override public String toString() { return "{" + departure + " -> " + destination + '}'; } } }

プログラムの結果

乗客{1-> 2}はストップ番号1で待っています
乗客{1-> 5}は、ストップ番号1で待っています
乗客{2-> 3}はストップ番号2で待っています
乗客{2-> 5}はストップ番号2で待っています
乗客{3-> 4}はストップ番号3で待っています
乗客{4-> 5}はストップ番号4で待っています
乗客{4-> 5}はストップ番号4で待っています
バスは公園を出発しました。
ストップナンバー1
乗客{1-> 5}がバスに乗り込んだ。
乗客{1-> 2}がバスに乗り込みました。
ストップナンバー2
乗客{2-> 3}がバスに乗り込んだ。
{1 -> 2} .
{2 -> 5} .
№ 3
{2 -> 3} .
{3 -> 4} .
№ 4
{4 -> 5} .
{3 -> 4} .
{4 -> 5} .
№ 5
{1 -> 5} .
{2 -> 5} .
{4 -> 5} .
{4 -> 5} .
.

ところで、フェイザー機能はCountDownLatchの動作を再現できます。

Phaserを使用したCountDownLatchの例

 import java.util.concurrent.Phaser; public class NewRace { private static final Phaser START = new Phaser(8); private static final int trackLength = 500000; public static void main(String[] args) throws InterruptedException { for (int i = 1; i <= 5; i++) { new Thread(new Car(i, (int) (Math.random() * 100 + 50))).start(); Thread.sleep(100); } while (START.getRegisteredParties() > 3) //,     Thread.sleep(100); //  .  ,  100ms Thread.sleep(100); System.out.println(" !"); START.arriveAndDeregister(); Thread.sleep(100); System.out.println("!"); START.arriveAndDeregister(); Thread.sleep(100); System.out.println("!"); START.arriveAndDeregister(); } public static class Car implements Runnable { private int carNumber; private int carSpeed; public Car(int carNumber, int carSpeed) { this.carNumber = carNumber; this.carSpeed = carSpeed; } @Override public void run() { try { System.out.printf(" №%d    .\n", carNumber); START.arriveAndDeregister(); START.awaitAdvance(0); Thread.sleep(trackLength / carSpeed); System.out.printf(" №%d !\n", carNumber); } catch (InterruptedException e) { } } } }

誰かが重宝してくれたら、とても嬉しいです=）

Phaserの詳細はこちら。
シンクロナイザーの詳細と例を参照するには、こちらをご覧ください。

java.util.concurrentの優れた概要はこちらをご覧ください。

シンクロナイザー参照java.util.concurrent。*

セマフォ

CountDownLatch

CyclicBarrier

エクスチェンジャー<V>

フェイザー

More articles: