1前書き

この記事では、


PriorityBlockingQueue


クラスに焦点を当て、いくつかの実用的な例を紹介します。



Queue


が何であるかを既に知っているという前提から始めて、最初に

要素の中の要素を説明します。

PriorityBlockingQueue

は優先順位

で並べられます。

その後、このタイプのキューを使用してスレッドをブロックする方法を説明します。

最後に、これら2つの機能を一緒に使用すると、複数のスレッドにまたがってデータを処理するときにどのように役立つかを示します。


2要素の優先順位

標準のキューとは異なり、__PriorityBlockingQueueに任意の種類の要素を追加することはできません。

  1. 実装する要素を追加する



同等


。条件に

Comparable

を実装していない要素を追加する



Comparator


も指定する必要があります。


Comparator

実装または

Comparable

実装のいずれかを使用して要素を比較することで、

PriorityBlockingQueue

は常に並べ替えられます。

目的は、最も優先順位の高い要素が常に最初に並ぶように比較ロジックを実装することです。次に、キューから要素を削除すると、それが常に最も優先順位の高い要素になります。

まず、キューを複数のスレッドで使用するのではなく、単一のスレッドで使用します。こうすることで、単体テストで要素がどのように順序付けられているかを簡単に証明できます。 

PriorityBlockingQueue<Integer> queue = new PriorityBlockingQueue<>();
ArrayList<Integer> polledElements = new ArrayList<>();

queue.add(1);
queue.add(5);
queue.add(2);
queue.add(3);
queue.add(4);

queue.drainTo(polledElements);

assertThat(polledElements).containsExactly(1, 2, 3, 4, 5);

ご覧のとおり、要素をランダムな順序でキューに追加したにもかかわらず、それらをポーリングし始めるとそれらは順序付けされます。これは、


Integer


クラスが__Comparableを実装しているためです。昇順にキューから取り出します。

  • 2つの要素が比較され、同じである場合、それらがどのように順序付けられるかについての保証はありません。


3

キュー

を使用してブロックする

標準的なキューを扱う場合は、

poll（）

を呼び出して要素を取得します。ただし、キューが空の場合は、

poll（）

を呼び出すと__nullが返されます。


PriorityBlockingQueue

は

BlockingQueue

インターフェースを実装しています。これにより、空のキューから削除するときにブロックすることができるようになります。

take（）

メソッドを使ってみましょう。 

PriorityBlockingQueue<Integer> queue = new PriorityBlockingQueue<>();

new Thread(() -> {
  System.out.println("Polling...");

  try {
      Integer poll = queue.take();
      System.out.println("Polled: " + poll);
  } catch (InterruptedException e) {
      e.printStackTrace();
  }
}).start();

Thread.sleep(TimeUnit.SECONDS.toMillis(5));
System.out.println("Adding to queue");
queue.add(1);


sleep（）

を使用することは物事をデモンストレーションするための少し脆い方法ですが、このコードを実行すると以下のようになります。 

Polling...
Adding to queue
Polled: 1

これは、アイテムが追加されるまで

take（）

がブロックされたことを証明します。

  1. スレッドは、開始したことを証明するために「Polling」を印刷します

  2. テストはそれからスレッドを証明するために、およそ5秒間休止します.

この時点で

take（）

を呼び出しているはずです
。私たちはキューに追加して、多かれ少なかれ即座に“ Polled:1”を見るべきです


take（）

が利用可能になるとすぐに要素を返したことを証明するため

また、

BlockingQueue

インターフェースは、フルキューに追加するときにブロックする方法も提供しています。

ただし、

PriorityBlockingQueue

は無制限です。これはそれがいっぱいになることは決してないということを意味しているので、新しい要素を追加することは常に可能です。


4ブロッキングと優先順位付けの併用

これで、__PriorityBlockingQueueの2つの重要な概念について説明しました。これらを両方使用しましょう。前の例を単純に拡張できますが、今回はキューに要素を追加します。 

Thread thread = new Thread(() -> {
    System.out.println("Polling...");
    while (true) {
        try {
            Integer poll = queue.take();
            System.out.println("Polled: " + poll);
        }
        catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
});

thread.start();

Thread.sleep(TimeUnit.SECONDS.toMillis(5));
System.out.println("Adding to queue");

queue.addAll(newArrayList(1, 5, 6, 1, 2, 6, 7));
Thread.sleep(TimeUnit.SECONDS.toMillis(1));

繰り返しますが、これは__sleep（）を使用しているため少し脆いですが、それでもまだ有効な使用例を示しています。これで、要素が追加されるのを待って、ブロックするキューができました。それから、一度にたくさんの要素を追加し、それらが優先順位に従って処理されることを示します。出力は次のようになります。 

Polling...
Adding to queue
Polled: 1
Polled: 1
Polled: 2
Polled: 5
Polled: 6
Polled: 6
Polled: 7


5結論

このガイドでは、いくつかのアイテムが追加されるまでスレッドをブロックするために

PriorityBlockingQueue

を使用する方法を説明しました。また、それらのアイテムを優先度に基づいて処理することもできます。

これらの例の実装はhttps://github.com/eugenp/tutorials/tree/master/core-java-concurrency-collections[GitHubに追加]をご覧ください。これはMavenベースのプロジェクトなので、そのまま実行するのは簡単なはずです。