1. 概要
このチュートリアルでは、Java でリングバッファを実装する方法を学びます。
リングバッファ
リングバッファ（または円形バッファ）は、2 つ以上のスレッド間でデータをバッファするために使われる境界のある円形のデータ構造です。 リングバッファに書き込みを続けると、最後に到達したときにラップアラウンドされます。 どのように動作するか
リングバッファは、境界で折り返される固定サイズの配列を使って実装されています。
配列とは別に、3 つのことを記録しています。
- 要素を挿入するためのバッファ内の次の空きスロット、
- バッファ内の次の未読要素、
- 配列の終端 – バッファが配列の先頭に回り込む点
リングバッファがこれらの要件を処理する方法の仕組みは実装によって異なります。 たとえば、Wikipedia のエントリでは、4 つのポインターを使用する方法が示されています。
ここでは、シーケンスを使用するリングバッファの Disruptor の実装からアプローチを借用します。
- Write Sequence: -1 から始まり、要素を挿入すると 1 ずつ増加します
- Read Sequence: 0 から始まり、要素を消費すると 1 ずつ増加します
配列をインデックスにマッピングするには mod 演算を使用します。
arrayIndex = sequence % capacity修飾演算は、シーケンスを境界でラップし、バッファ内のスロットを作成します。
要素を挿入する方法を見てみましょう。
要素を消費するには、ポストインクリメントを行います:
element = bufferこの場合、シーケンスに対してポストインクリメントを行っています。 要素を消費してもバッファから削除されることはなく、上書きされるまで配列に残ります。 空のバッファと満杯のバッファ
配列の周りを回るとき、バッファ内のデータの上書きを開始します。 バッファが満杯の場合、リーダーが消費したかどうかに関係なく最も古いデータを上書きするか、または読まれていないデータを上書きしないようにするかを選択できます。
リーダーが中間値または古い値を見逃す余裕がある場合（たとえば、株価ティッカー）、データが消費されるのを待たずにデータを上書きすることが可能です。 一方、読者がすべての値を消費しなければならない場合（電子商取引など）、バッファに空きスロットができるまで待つ（ブロック／ビジーウェイト）必要があります。
バッファのサイズが未読要素の数に等しい場合、バッファは満杯です:
size = (writeSequence - readSequence) + 1isFull = (size == capacity) 
書き込みシーケンスが読み込みシーケンスより遅れている場合、バッファは空です:
isEmpty = writeSequence < readSequence
バッファが空の場合は、NULL 値を返します:
2.2。 利点と欠点
リングバッファは、効率的なFIFOバッファです。 前もって割り当てることができる固定サイズの配列を使用し、効率的なメモリ・アクセス・パターンを可能にします。 要素のシフトを必要としないため、要素の消費を含め、すべてのバッファ操作は一定時間 O(1) です。
その反面、リングバッファの正しいサイズを決定することは重要です。 たとえば、バッファのサイズが小さく、読み取りが遅い場合、書き込み操作が長時間ブロックされる可能性があります。 動的なサイズ設定を使用することもできますが、データを移動する必要があり、前述の利点のほとんどを失うことになります。
Java
リングバッファの仕組みを理解したところで、Java で実装してみましょう。 初期化
最初に、バッファを定義済みの容量で初期化するコンストラクタを定義します。
public CircularBuffer(int capacity) { this.capacity = (capacity < 1) ? DEFAULT_CAPACITY : capacity; this.data = (E) new Object; this.readSequence = 0; this.writeSequence = -1;}これは空のバッファを作成し、前のセクションで説明したようにシーケンスフィールドを初期化します。 Offer
次に、次の利用可能なスロットでバッファに要素を挿入し、成功時にtrueを返すオファー操作を実装します。 バッファが空のスロットを見つけられない場合、つまり、未読の値を上書きできない場合は false を返します。
Offer メソッドを Java で実装しましょう。
public boolean offer(E element) { boolean isFull = (writeSequence - readSequence) + 1 == capacity; if (!isFull) { int nextWriteSeq = writeSequence + 1; data = element; writeSequence++; return true; } return false;}そこで、書き込みシーケンスを増分し、次の利用可能なスロットに対する配列内のインデックスを計算します。
では、書き込みシーケンスをインクリメントし、次の利用可能なスロットの配列のインデックスを計算します。その後、データをバッファに書き込み、更新された書き込みシーケンスを格納します。 Poll
最後に、次の未読要素を取得し削除するポーリング操作を実装します。
public E poll() { boolean isEmpty = writeSequence < readSequence; if (!isEmpty) { E nextValue = data; readSequence++; return nextValue; } return null;}ここで、配列のインデックスを計算して、現在の読み取りシーケンスでデータを読み取っています。
@Testpublic void givenCircularBuffer_whenAnElementIsDequeued_thenElementMatchesEnqueuedElement() { CircularBuffer buffer = new CircularBuffer<>(defaultCapacity); buffer.offer("Triangle"); String shape = buffer.poll(); assertEquals("Triangle", shape);}Producer-Consumer Problem
2 つ以上のスレッド間でデータを交換するためにリングバッファを使用することについてお話しましたが、これは Producer-Consumer 問題と呼ばれる同期問題の一例です。 Java では、セマフォ、境界付きキュー、リングバッファなどを使用して、さまざまな方法でプロデューサー-コンシューマー問題を解決できます。
リングバッファに基づく解決策を実装してみましょう。 2293>
プロデューサーはバッファにデータを書き込み、writeSequence をインクリメントし、コンシューマはバッファから読み取り、readSequence をインクリメントするだけです。 2293>
しかし、消費者が writeSequence フィールドの最新値を見ることができ (可視性)、データがバッファで実際に利用可能になる前に writeSequence が更新されない (順序付け) ことを保証する必要があります。
この場合、シーケンスフィールドをvolatileにすることで、リングバッファをコンカレントかつロックフリーにできる：
private volatile int writeSequence = -1, readSequence = 0;オファーメソッドでは、volatileフィールドwriteSequenceへの書き込みは、シーケンスを更新する前にバッファへの書き込みが起こることを保証している。 同時に、volatileの可視性保証は、消費者が常にwriteSequenceの最新の値を見ることができることを保証している。 Producer
リングバッファに書き込む単純な producer Runnable を実装しましょう。
public void run() { for (int i = 0; i < items.length;) { if (buffer.offer(items)) { System.out.println("Produced: " + items); i++; } }}プロデューサースレッドは、ループ内の空のスロットで待機します (ビジーウェイト)。 Consumer
バッファから読み取る consumer Callable を実装します。
public T call() { T items = (T) new Object; for (int i = 0; i < items.length;) { T item = buffer.poll(); if (item != null) { items = item; System.out.println("Consumed: " + item); } } return items;}バッファから null 値を受信すると、consumer スレッドは印刷せずに続行します。
ドライバ コードを記述します。
executorService.submit(new Thread(new Producer<String>(buffer)));executorService.submit(new Thread(new Consumer<String>(buffer)));プロデューサー/コンシューマー プログラムを実行すると、次のような出力が得られます：
Produced: CircleProduced: Triangle Consumed: CircleProduced: Rectangle Consumed: Triangle Consumed: RectangleProduced: SquareProduced: Rhombus Consumed: SquareProduced: Trapezoid Consumed: Rhombus Consumed: TrapezoidProduced: PentagonProduced: PentagramProduced: Hexagon Consumed: Pentagon Consumed: PentagramProduced: Hexagram Consumed: Hexagon Consumed: Hexagram結論
このチュートリアルでは、リングバッファを実装する方法を学び、プロデューサーとコンシューマーの問題の解決に使用できる方法を検討しました。