1概要

このチュートリアルでは、https://github.com/JCTools/JCTools[JCTools]（Java Concurrency Tools）ライブラリを紹介します。

簡単に言えば、これはマルチスレッド環境での作業に適した多数のユーティリティデータ構造を提供します。


2ノンブロッキングアルゴリズム

  • 伝統的に、変更可能な共有状態で動作するマルチスレッドコードは、データの一貫性とパブリケーション（あるスレッドによって行われた変更が別のスレッドに見える）を保証するためにロック** を使用します。

この方法にはいくつかの欠点があります。

  • ロックを獲得しようとしてスレッドがブロックされる可能性があります。

他のスレッドの操作が終了するまでは何も進行しません – これは
並列処理を効果的に防ぐ
** ロック競合が激しいほど、JVMの処理に費やす時間が長くなります。

スレッドのスケジューリング、競合と待ち行列の管理
スレッドとそれが実際に行っている仕事が少ない
** 複数のロックが関係している場合はデッドロックが発生する可能性があります。

誤った順序で取得/解放された
**

優先順位の逆転

ハザードがある可能性があります – 優先度の高いスレッドがロックしようとしてロックされています
優先度の低いスレッドが保持しているロックを取得する
** ほとんどの場合、粒度の粗いロックが使用され、並列処理が損なわれます。

ロット – きめ細かいロックは、より慎重な設計を必要とし、ロックのオーバーヘッドを増加させ、そしてエラーを起こしやすい

代替案は、

非ブロッキングアルゴリズム、すなわち、任意のスレッドの失敗または中断が別のスレッド

の失敗または中断を引き起こすことができないアルゴリズムを使用することである。

関与するスレッドのうちの少なくとも１つが任意の期間にわたって進行することが保証されている場合、すなわち処理中にデッドロックが生じ得ない場合、ノンブロッキングアルゴリズムは「ロックフリー」である。

さらに、スレッドごとの進捗が保証されている場合、これらのアルゴリズムは「待つ必要がありません」。

これは、優れたhttps://www.amazon.com/Java-Concurrency-Practice-Brian-Goetz/dp/0321349601/ref=sr

1

1[Java Concurrency in Practice]の本からの非ブロックの

Stack

の例です。それは基本的な状態を定義します。 

public class ConcurrentStack<E> {

    AtomicReference<Node<E>> top = new AtomicReference<Node<E>>();

    private static class Node <E> {
        public E item;
        public Node<E> next;

       //standard constructor
    }
}

また、APIメソッドもいくつかあります。 

public void push(E item){
    Node<E> newHead = new Node<E>(item);
    Node<E> oldHead;

    do {
        oldHead = top.get();
        newHead.next = oldHead;
    } while(!top.compareAndSet(oldHead, newHead));
}

public E pop() {
    Node<E> oldHead;
    Node<E> newHead;
    do {
        oldHead = top.get();
        if (oldHead == null) {
            return null;
        }
        newHead = oldHead.next;
    } while (!top.compareAndSet(oldHead, newHead));

    return oldHead.item;
}

このアルゴリズムは、きめ細かいcompare-and-swap（https://en.wikipedia.org/wiki/Compare-and-swap[CAS]）命令を使用しており、

lock-free

です（複数のスレッドが

topを呼び出した場合でも）。 .compareAndSet（）

同時に、そのうちの1つが成功することが保証されていますが、CASが最終的に特定のスレッドで成功するという保証はないため、

wait-free

はできません。


3依存

まず、JCTools依存関係を

pom.xml

に追加しましょう。 

<dependency>
    <groupId>org.jctools</groupId>
    <artifactId>jctools-core</artifactId>
    <version>2.1.2</version>
</dependency>

利用可能な最新バージョンはhttps://search.maven.org/classic/#search%7C1%7Cg%3A%22org.jctools%22%20AND%20a%3A%22jctools-core%22で入手可能です。[Maven Central]


4 JCToolsキュー

このライブラリは、マルチスレッド環境で使用するための多数のキューを提供します。つまり、1つ以上のスレッドがキューに書き込み、1つ以上のスレッドがスレッドセーフなロックフリーの方法でキューから読み取ります。

すべての

Queue

実装の共通インターフェースは

org.jctools.queues.MessagePassingQueue

です。


4.1. キューの種類

すべてのキューは、生産者/消費者ポリシーに基づいて分類できます。


  • 単一の生産者、単一の消費者 –

    このようなクラスは、

接頭辞

Spsc

、例:

SpscArrayQueue



単一のプロデューサ、複数のコンシューマ – **

Spmc

プレフィックスを使用、

例えば

SpmcArrayQueue



複数のプロデューサ、単一のコンシューマ – **

Mpsc

プレフィックスを使用


MpscArrayQueue



複数のプロデューサー、複数のコンシューマー – **

Mpmc

プレフィックスを使用、

例えば

MpmcArrayQueue

内部ポリシーのチェックはありません。

誤った使用法の場合、キューは黙って誤動作する可能性があります** 。

例えば。以下のテストは、2つのスレッドから

single-producer

キューを作成し、コンシューマが異なるプロデューサからのデータを見ることが保証されていなくてもパスします。 

SpscArrayQueue<Integer> queue = new SpscArrayQueue<>(2);

Thread producer1 = new Thread(() -> queue.offer(1));
producer1.start();
producer1.join();

Thread producer2 = new Thread(() -> queue.offer(2));
producer2.start();
producer2.join();

Set<Integer> fromQueue = new HashSet<>();
Thread consumer = new Thread(() -> queue.drain(fromQueue::add));
consumer.start();
consumer.join();

assertThat(fromQueue).containsOnly(1, 2);

4.2. キュー実装

上記の分類を要約すると、以下はJCToolsキューのリストです。



  • SpscArrayQueue

    –

    シングルプロデューサ、シングルコンシューマ、配列を使用

内部的に、限界容量



SpscLinkedQueue

– ** シングルプロデューサ、シングルコンシューマ、リンクの使用

内部的にリストし、未結合容量



SpscChunkedArrayQueue

– ** シングルプロデューサ、シングルコンシューマ、開始

初期容量で最大容量まで成長



SpscGrowableArrayQueue

– ** シングルプロデューサ、シングルコンシューマ、開始

初期容量で最大容量まで成長します。これは同じです

SpscChunkedArrayQueue

としての契約、唯一の違いは内部的です
チャンク管理。

SpscChunkedArrayQueue

を使用することをお勧めします
実装が単純化されているため



SpscUnboundedArrayQueue

– ** シングルプロデューサ、シングルコンシューマ、用途

内部的に配列された、未結合容量



SpmcArrayQueue

– ** 単一のプロデューサ、複数のコンシューマが、

内部配列、バウンド容量



MpscArrayQueue

– ** 複数のプロデューサ、単一のコンシューマ、

内部配列、バウンド容量



MpscLinkedQueue

– ** 複数のプロデューサ、単一のコンシューマ、

内部的にリンクリスト、未結合容量



MpmcArrayQueue

– ** 複数のプロデューサ、複数のコンシューマが、

内部配列、バウンド容量


4.3. アトミックキュー

前のセクションで説明したすべてのキューは


sun.misc.Unsafe


を使用します。ただし、Java 9とhttp://openjdk.java.net/jeps/260[JEP-260]の出現により、このAPIはデフォルトではアクセスできないようになりました。

そのため、

sun.misc.Unsafe

の代わりに

java.util.concurrent.atomic.AtomicLongFieldUpdater

（public API、less performant）を使用する代替キューがあります。

それらは上記のキューから生成され、それらの名前の間に

Atomic

という単語が挿入されます。

SpscChunkedAtomicArrayQueue

または

MpmcAtomicArrayQueue

。

HotSpot Java9やJRockitのように

sun.misc.Unsafe

が禁止されているか無効である環境でのみ、可能であれば「通常の」キューを使用し、

AtomicQueues

を使用することをお勧めします。


4.4. 容量

すべてのJCToolsキューにも最大容量があるか、バインドされていない可能性があります。

キューがいっぱいになり、容量によって制限されると、新しい要素の受け入れを停止します。

次の例では、

  • キューをいっぱいにする

  • その後、新しい要素を受け入れないようにする

  • そこから排水し、要素を追加することが可能であることを確認する

その後

読みやすくするために、いくつかのコードステートメントは削除されています。完全な実装はhttps://github.com/eugenp/tutorials/blob/master/libraries/src/test/java/com/baeldung/jctools/JCToolsUnitTest.java#L45[on GitHub]にあります。 

SpscChunkedArrayQueue<Integer> queue = new SpscChunkedArrayQueue<>(8, 16);
CountDownLatch startConsuming = new CountDownLatch(1);
CountDownLatch awakeProducer = new CountDownLatch(1);

Thread producer = new Thread(() -> {
    IntStream.range(0, queue.capacity()).forEach(i -> {
        assertThat(queue.offer(i)).isTrue();
    });
    assertThat(queue.offer(queue.capacity())).isFalse();
    startConsuming.countDown();
    awakeProducer.await();
    assertThat(queue.offer(queue.capacity())).isTrue();
});

producer.start();
startConsuming.await();

Set<Integer> fromQueue = new HashSet<>();
queue.drain(fromQueue::add);
awakeProducer.countDown();
producer.join();
queue.drain(fromQueue::add);

assertThat(fromQueue).containsAll(
  IntStream.range(0, 17).boxed().collect(toSet()));


5その他のJCToolsのデータ構造

JCToolsは、Queue以外のデータ構造もいくつか提供しています。

それらすべてが以下にリストされています。



  • NonBlockingHashMap

    –

    ロックフリーの

    ConcurrentHashMap

    代替

より良いスケーリング特性と一般的により低い突然変異コストで。それは

sun.misc.Unsafe

を介して実装されているため、これを使用することはお勧めできません。
HotSpot Java 9またはJRockit環境のクラス



NonBlockingHashMapLong

– **

NonBlockingHashMap

に似ていますが使用します

プリミティブ

長

キー



NonBlockingHashSet

– ** シンプルなラッパー

JDKのように

__NonBlockingHashMapについて







java.util.Collections.newSetFromMap（）




NonBlockingIdentityHashMap

– **

NonBlockingHashMap__と似ていますが

アイデンティティによってキーを比較します。




  • NonBlockingSetInt

    –


    実装されたマルチスレッドのビットベクトルセット

プリミティブ

longs

の配列として。サイレントオートボクシングの場合は無効に機能します。


6. 性能試験

JDKの

ArrayBlockingQueue

とJCToolsのキューのパフォーマンスを比較するためにhttp://openjdk.java.net/projects/code-tools/jmh/[JMH]を使用しましょう。 JMHはSun/Oracle JVMの達人からのオープンソースのマイクロベンチマークフレームワークで、コンパイラ/jvm最適化アルゴリズムの不確定性から私たちを守ります）。詳しくは

この記事

をご覧ください。

以下のコードスニペットは、読みやすさを向上させるためにいくつかのステートメントを見逃していることに注意してください。

GitHub:

で完全なソースコードを見つけてください。 

public class MpmcBenchmark {

    @Param({PARAM__UNSAFE, PARAM__AFU, PARAM__JDK})
    public volatile String implementation;

    public volatile Queue<Long> queue;

    @Benchmark
    @Group(GROUP__NAME)
    @GroupThreads(PRODUCER__THREADS__NUMBER)
    public void write(Control control) {
       //noinspection StatementWithEmptyBody
        while (!control.stopMeasurement && !queue.offer(1L)) {
           //intentionally left blank
        }
    }

    @Benchmark
    @Group(GROUP__NAME)
    @GroupThreads(CONSUMER__THREADS__NUMBER)
    public void read(Control control) {
       //noinspection StatementWithEmptyBody
        while (!control.stopMeasurement && queue.poll() == null) {
           //intentionally left blank
        }
    }
}

結果（95パーセンタイルの抜粋、1操作あたりのナノ秒）: 

MpmcBenchmark.MyGroup:MyGroup·p0.95 MpmcArrayQueue sample 1052.000 ns/op
MpmcBenchmark.MyGroup:MyGroup·p0.95 MpmcAtomicArrayQueue sample 1106.000 ns/op
MpmcBenchmark.MyGroup:MyGroup·p0.95 ArrayBlockingQueue sample 2364.000 ns/op



  • MpmcArrayQueue

    のパフォーマンスが

    MpmcAtomicArrayQueue

    および

    ArrayBlockingQueue

    のパフォーマンスよりもわずかに2倍遅いことがわかります。

    **

7. JCToolsを使用することの欠点

たとえば、大規模で成熟したプロジェクトで

MpscArrayQueue

を使い始めるときの状況を考えてみましょう（単一のコンシューマが存在しなければならないことに注意してください）。

残念ながら、プロジェクトが大きいため、誰かがプログラミングまたは設定エラーを起こし、キューが複数のスレッドから読み取られる可能性があります。このシステムは以前と同じように動作しているように見えますが、今では消費者がいくつかのメッセージを見逃す可能性があります。これは大きな影響を与える可能性があり、デバッグが非常に難しい実際の問題です。

理想的には、JCToolsにスレッドアクセスポリシーを保証させる特定のシステムプロパティでシステムを実行することが可能であるべきです。例えば。

ローカル/テスト/ステージング環境（ただし本番環境ではない）では有効になっている可能性があります。残念ながら、JCToolsはそのようなプロパティを提供していません。

もう1つ考慮すべきことは、JCToolsがJDKの対応するものよりもかなり速いことを保証していても、カスタムキュー実装を使い始めるのと同じ速度でアプリケーションが得られるわけではないことです。ほとんどのアプリケーションは、スレッド間で多くのオブジェクトを交換することはなく、ほとんどの場合I/Oに制限があります。


8結論

これで、JCToolsが提供するユーティリティクラスの基本的な理解が深まり、高負荷の下でのJDKの対応クラスと比較して、それらがどの程度うまく機能するかがわかりました。

結論として、** スレッド間で多数のオブジェクトを交換する場合にのみライブラリを使用する価値があります。その場合でも、スレッドアクセスポリシーを維持するためには細心の注意を払う必要があります。

いつものように、上記のサンプルの完全なソースコードはhttps://github.com/eugenp/tutorials/tree/master/libraries[over on GitHub]にあります。