プロデューサ - コンシューマ間スレッド間通信

Question

スレッド間通信に問題があり、 "ダミーメッセージ"を全面的に使用して解決しました。これは悪い考えですか？可能な解決策は何ですか？

例問題があります。

メインスレッドは、レコードを処理してデータベースに挿入するスレッドを開始します。 メインスレッドは、おそらく巨大なファイルを読み取り、1つのレコード（オブジェクト）を次々にブロッキングキューに格納します。処理スレッドはキューから読み込み、動作します。

「スレッドの処理」にはどのように停止する必要がありますか？ キューは空でもかまいませんが、作業は完了せず、スレッドの処理が完了してもそれを中断することはできません。

だから、処理スレッドは挿入が同じトランザクションおよびトランザクションでなければならないこと

while (queue.size() > 0 || !Thread.currentThread().isInterrupted()) { 
    MyObject object= queue.poll(100, TimeUnit.MILLISECONDS); 
    if (object != null) { 
     String data = object.getData(); 
     if (data.equals("END")) { 
      break; 
     } 
    // do work 
    } 
} 
// clean-up 
synchronized queue) { 
    queue.notifyAll(); 
} 
return; 

とメインスレッド

// ...start processing thread... 
while(reader.hasNext(){ 
    // ...read whole file and put data in queue... 
} 
MyObject dummy = new MyObject(); 
dummy.setData("END"); 
queue.put(dummy); 
//Note: empty queue here means work is done 
while (queue.size() > 0) { 
    synchronized (queue) { 
     queue.wait(500); // over-cautios locking prevention i guess 
    } 
} 

注メインスレッドで を扱うことができないん。

これを行うにはどうすればよいでしょうか？ （私は学んでおり、「間違ったやり方」を始めたくない）

Answer 1

これらのダミーメッセージは有効です。それは "毒"と呼ばれています。プロデューサーが消費者にそれを止めるために送るもの。

他の可能性は、メインスレッドのどこかでThread.interrupt（）を呼び出して、それに応じてInterruptedExceptionをキャッチして処理することです。

Answer 2

「ダミーメッセージ」を使用して完全に解決しました。これは の悪い考えですか？可能な解決策は何ですか？

これは「Poison Pills」と呼ばれ、スレッドベースのサービスを停止するための妥当な方法です。

ただし、プロデューサとコンシューマの数がわかっている場合にのみ動作します。

あなたが投稿したコードには、データを生成する「メインスレッド」とデータを消費する「処理スレッド」の2つのスレッドがあり、この状況では「ポイズンピル」がうまく機能します。

しかし、他の生産者もいる場合、消費者はいつ停止するのか（すべての生産者が "Poison Pills"を送信する場合のみ）を知っているか、すべての生産者の数を正確に把握し、消費者の「ポイズンピル」の数がプロデューサーの数に等しい場合は、すべてのプロデューサーが機能しなくなり、消費者が停止することを意味します。

"メインスレッド"ではInterruptedExceptionをキャッチする必要があります。そうでない場合は、 "メインスレッド"が "ポイズンピル"を設定できない可能性があります。あなたはすべてのあなたの問題を解決するためにExecutorServiceを使用しようとすることができ、また、以下のように

... 
try { 
    // do normal processing 
} catch (InterruptedException e) { /* fall through */ } 
finally { 
    MyObject dummy = new MyObject(); 
    dummy.setData("END"); 
    ... 
} 
... 

を行うことができます。

（あなただけのいくつかの作品を行い、その後、すべてが終了したときに停止する必要があるとき、それは動作します）

void doWorks(Set<String> works, long timeout, TimeUnit unit) 
    throws InterruptedException { 
    ExecutorService exec = Executors.newCachedThreadPool(); 
    try { 
     for (final String work : works) 
      exec.execute(new Runnable() { 
        public void run() { 
         ... 
        } 
       }); 
    } finally { 
     exec.shutdown(); 
     exec.awaitTermination(timeout, unit); 
    } 
} 

私が学んだし、「それを間違った方法をやって」を開始する必要はありません

書籍：Java Concurrency in Practiceを読む必要があります。私を信じて、それは最高です。

Answer 3

私が最近のプロジェクトで行ったことは、キューをラップしてから'isOpen()'メソッドを追加することです。

while (reader.hasNext()) { 
    // read the file and put it into the queue 
    dataQ.put(someObject); 
} 
// now we're done 
dataQ.close(); 

と読者のスレッド：：

class ClosableQ<T> { 

    boolean isOpen = true; 

    private LinkedBlockingQueue<T> lbq = new LinkedBlockingQueue<T>(); 

    public void put(T someObject) { 
     if (isOpen) { 
     lbq.put(someObject); 
     } 
    } 

    public T get() { 
     if (isOpen) { 
     return lbq.get(0); 
     } 
    } 

    public boolean isOpen() { 
     return isOpen; 
    } 

    public void open() { 
     isOpen = true; 
    } 

    public void close() { 
     isOpen = false; 
    } 
} 

だからあなたのライター・スレッドのようなものになり

while (dataQ.isOpen) { 
    someObject = dataQ.get(); 
} 

もちろんの代わりにリストを拡張することができますが、それはユーザーにレベルを与えますあなたが望むかもしれないアクセス。 AtomicBooleanのように、このコードにいくつかの並行処理を追加する必要があります。