JAVA開発における実際の状態の依存関係管理のためのブロッキングキューの実装

クラス ライブラリ自体には、状態の依存関係を持つ多くのクラスが含まれています。 FutureTask、BlockingQueue など。これらのクラスの一部の操作は、状態の前提条件に基づいています。たとえば、空のキューから要素を削除したり、未完了のタスクの計算結果を取得したりすることはできません。これら 2 つの操作を実行する前に、キューが空ではない状態になるか、タスクが完了状態になるまで待つ必要があります。状態依存クラスを作成する最も簡単な方法は、クラス ライブラリに基づいてクラスを構築することです。ただし、クラス ライブラリに必要な機能がない場合は、Java 言語とクラス ライブラリによって提供される基礎となるメカニズムを使用して、独自の同期メカニズムを構築することもできます。

そこで、この記事では、独自の状態依存関係クラスを構築する方法を紹介します。プロセスと最終結果を得る方法を理解するために、最も単純な構築から複雑な標準化された構築まで段階的に紹介します。

状態の依存関係の管理

ブロック可能な状態の依存関係の操作は、次の疑似コードに示されています。

acquire lock on object state //首先获取锁
     while (precondition does not hold) { //前提条件是否满足，不满足则一直循环重试
        release lock //释放锁
        wait until precondition might hold  //等待知道满足前提条件
        optionally fail if interrupted or timeout expire //中断或者超时，各种异常
        reacquire lock //重新获取锁
    }

perform action //执行任务
release lock  //释放锁

ロックを取得し、条件が満たされているかどうかを確認します。満たされていない場合は、ロックを解放し、条件が満たされるまでブロック状態に入ります。または中断またはタイムアウトしました。待ってからロックを再取得します。タスクを実行してロックを解除します。

今この疑似コードを見ても直感的に理解できないかもしれませんが、この記事を読み進めれば、すべての操作がこの疑似コード アーキテクチャによって構築されていることがわかります。

ArrayBlockingQueue は、put と take の 2 つの操作を提供する境界付きキャッシュです。 これらにはすべて、要素を完全なキャッシュに入れることはできず、空のキャッシュから要素を取得することはできないという前提条件が含まれています。さて、私たちの目標は、そのような ArrayBlockingQueue を構築することです。

次に、前提条件が満たされない状況に対処するために異なる方法を使用する、境界付きキャッシュの 2 つの実装を紹介します。

まず、次の基本クラス BaseBoundeBuffer を見てみましょう。後の実装ではこの基本クラスが拡張されます。これは配列ベースの循環キャッシュであり、含まれる変数 buf、head、tail、count はキャッシュの組み込みロックによって保護されます。また、同期 doPut および doTake メソッドも提供しており、サブクラスでは、put および take 操作がこれらのメソッドを通じて実装され、基になる状態はサブクラスから隠蔽されます。

public abstract class BaseBoundedBuffer<V> {

    private final V[] buf;    
    private int tail;    
    private int head;    
    private int count;    
    protected BaseBoundedBuffer(int capacity) {        
    this.buf = (V[]) new Object[capacity];
        count = 0;
    }    
    protected synchronized final void doPut(V v) {
        buf[tail] = v;        if(++tail == buf.length)
            tail = 0;
        ++count;
    }    
    protected synchronized final V doTake() {
        V v = buf[head];
        buf[head] = null;        
        if(++head == buf.length)
        head = 0;
        --count;        
        return v;
    }    
    public synchronized final boolean isFull() {        
        return count == buf.length;
    }    
    public synchronized final boolean isEmpty() {        
        return count == 0;
    }
}

境界キャッシュの最初の実装では、put メソッドと take メソッドの両方を同期し、最初にチェックしてから実行し、失敗した場合は例外をスローします。

public class GrumpyBoundedBuffer<V> extends BaseBoundedBuffer{

    protected GrumpyBoundedBuffer(int capacity) {        
        super(capacity);
    }    
    public synchronized void put(V v) throws BufferFullException {        
        if(isFull()) {            
            throw new BufferFullException();
       }
        doPut(v);
    }    
    public synchronized V take() throws BufferFullException {        
        if(isEmpty())            
            throw new BufferFullException();        
            return (V) doTake();
    }
}

上記のように、前提条件が満たされていない場合、ここでのいわゆる例外は、キャッシュがいっぱいまたは空であることを指します。実際、この例外はプログラムが間違っていることを意味するものではありません。たとえば、赤信号が見えても信号機が異常であることを意味するのではなく、青信号が道路を横断するまで待っていることを意味します。したがって、これが意味するのは、呼び出し元は例外をキャッチして各キャッシュ操作を再試行する必要があるということです。

次のクライアント呼び出しコードを直接見てみましょう:

private static GrumpyBoundedBuffer gbb = new GrumpyBoundedBuffer(5);
...while(true) {    try {
        V item = gbb.take();        
        break;
    } catch(BufferEmptyException e) {
        Thread.sleep(500);
    }
}

端的に言うと、前提条件が満たされていない場合は、条件が満たされるまで再試行することで、ブロック効果が得られるように見えます。ただし、この場合、呼び出し元は前提条件の失敗を自分で処理する必要があり、CPU は常に占有されます。ここでの問題は、呼び出し側がこのキューを使用するのが非常に面倒になることです。

2 番目のメソッド SleepyBoundedBuffer は、ポーリングとスリープを通じて単純なブロック再試行メカニズムを実装し、呼び出し元が再試行メカニズムを取り除き、キャッシュの使用を簡素化できるようにします。以下のコードリストを見てください:

public class SleepyBoundedBuffer<V> extends BaseBoundedBuffer{

    protected SleepyBoundedBuffer(int capacity) {        
        super(capacity);        
        // TODO Auto-generated constructor stub
    }    
    public void put(V v) throws InterruptedException {        
        while(true) {            
            synchronized(this) {                
                if(!isFull()) {
                    doPut(v);                    
                    return;
                }
            }
            Thread.sleep(200);
        }
    }    
    public V take() throws InterruptedException{        
        while(true) {            
            synchronized(this) {                
            if(!isEmpty()) {                    
            return (V) doTake();
                }
            }
            Thread.sleep(200);
        }
    }
}

呼び出し側の観点から見ると、このアプローチは非常にうまく機能します。操作が前提条件を満たしている場合はすぐに実行され、そうでない場合はブロックされます。呼び出し元は失敗や再試行を処理する必要はありませんが、InterruptedException を処理する必要があります。ほとんどの適切に動作するブロッキング ライブラリ メソッドと同様に、SleepyBoundedBuffer は割り込みによるキャンセルをサポートします。

SleepyBoundedBuffer の問題は、スリープ時間の設定はどれくらいが妥当なのかということです。最適なパフォーマンスを達成するにはどうすればよいでしょうか?次の図に示すように、スレッド B は条件を true に設定しますが、この時点では A はまだスリープ状態であり、このスリープがパフォーマンスのボトルネックとなります。

それでは、条件が true の場合、スレッドがすぐに起動して実行されるようにする方法はあるのでしょうか?
それを大局的に見てみましょう。次の記事で説明します。

以上がJAVA開発における実際の状態の依存関係管理のためのブロッキングキューの実装の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。