0

私は生産者消費者プログラムを作りました。これは、GUI (Swing または SWT) のないコア Java の単なるプログラムです。オブジェクトをキューに入れるプロデューサーが 1 つあります。また、その共有キューのすべてのオブジェクトにスタッフ (文字列など) を追加する必要があるコンシューマーもいくつかあります。したがって、すべてのコンシューマーは、共有キュー内のすべてのオブジェクトを処理する必要があります。この場合、すべての BookShelf には、「books」ArrayList 内のすべてのコンシューマーからのアイテムが必要です。消費者。

質問:スレッドを正しく終了するには、コンシューマーでどの条件を使用する必要がありますか? プログラムのコードの一部を次に示します。多分私は間違った方法でそれを実装しました。

キューのオブジェクトは次のとおりです。

public class BookShelf {
private int id;
private String name;
private int height;
private int weigh;
List<String> books = Collections.synchronizedList(new ArrayList<String>());

public BookShelf(int id, String name) {
    this.id = id;
    this.name = name;
}
public void addBook(String book) {
  books.add(book);
}
public boolean eq(String book) {
synchronized (books) {
    for (String b: books) {
    if (b.equalsIgnoreCase(book)) {
        return true;
    }
    }
}
return false;
}
 other setters and getters..

}

プロデューサークラスは次のとおりです。

public class Producer implements Runnable {
private BlockingQueue myQueue;

public Producer(BlockingQueue myQueue) {
this.myQueue = myQueue;
}

public void run() {
for(int i=0; i<7; i++){
    try {
    System.out.println("Produced: " + i);
    BookShelf myBookShelf = new BookShelf(i, "book #" + i);
    myQueue.put(myBookShelf);
    } catch (InterruptedException ex) {
    //Proper handle
    }
}
}

}

これは消費者クラスの1つです：

 public class Consumer implements Runnable {
 private BlockingQueue myQueue;

public Consumer(BlockingQueue myQueue) {
    this.myQueue = myQueue; }

public void run() {
    while(true){
        try {
            BookShelf tempBookShelf = (BookShelf) myQueue.take();

            //eq() is my method to check if ArraList has a book.
            if (tempBookShelf.eq("Abc book")) {
                System.out.println("It already has book");
                myQueue.put(tempBookShelf);
                Thread.sleep(2000);
            } else {
                tempBookShelf.addBook("Abc book");
                myQueue.put(tempBookShelf);
                Thread.sleep(2000);
            }
        } catch (InterruptedException ex) {
            //Proper handle
        }
    }
}
}

メインクラスは次のとおりです。

public class ProducerConsumerTest {

public static void main(String[] args) {

     BlockingQueue sharedQueue = new LinkedBlockingQueue();
     Thread prodThread = new Thread(new Producer(sharedQueue));
     Thread consThread = new Thread(new Consumer(sharedQueue));
     Thread consThread2 = new Thread(new Consumer2(sharedQueue));

     prodThread.start();
     consThread.start();
     consThread2.start();
}
 }
4

4 に答える 4

1

各コンシューマーをプロデューサーに登録します。各コンシューマーには独自のキューがあり、プロデューサーはオブジェクトをすべてのキューに入れます。各コンシューマーは、オブジェクトの同じインスタンスを処理します。

    public interface Consumer{
        public void process(BookShelf bs);
    }

    public class Producer implements Runnable{
        private final List<Consumer> consumers = new CopyOnWriteArrayList<Consumer>(); // thread safe but not efficient with lots of changes

        public void register(Consumer c){
            consumers.add(c); // thread safe
        }

        public void run(){
            for(;;){
                BookShelf bs = generateBookShelfByWhateverMeans();
                for (Consumer c : consumers){
                    c.process(bs);
                }
            }
        }
    }

    public class BookShelfConsumer implements Runnable, Consumer{
        private final BlockingQueue<BookShelf> queue = new LinkedTransferQueue<BookShelf>(); // unbounded & thread safe

        public void process(BookShelf bs){
            queue.offer(bs); // non-blocking
        }

        public void run(){
            for(;;){
                BookShelf bs = queue.take(); // blocks until got object or interrupted 
                // catch InterruptedException 
                // do whatever this consumer is supposed to do with the object
            }
        }
    }
于 2013-04-13T09:57:27.567 に答える
0

同期されたリストの代わりに Bookshelf でConcurrentLinkedQueueを使用することをお勧めします。これはロックフリーで (同期する必要はありません)、おそらくより効率的です。

コンシューマーを終了するには、while(true)ループをwhile(!cancel)ループに変更します。cancelfalse に初期化するインスタンス変数として各コンシューマーにブール値を与え、truecancel()に設定するメソッドをそれらに与えcancelます。顧客とのやり取りが終わったら、顧客を呼びcancel()ましょう。常にすべてのコンシューマーを一度にキャンセルする場合 (一部を選択的にキャンセルするのではなく) cancel、 instance の代わりにstatic を使用できますcancel。

于 2013-04-13T13:49:15.813 に答える
0

代わりにSwingWorkerを使用してみます。終了時に実行されるdone()メソッドがあります。コード例については、このページを参照してください。

使用しているのが Swing でない場合は、Swt に Jobs という同様の機能があります。例については、このページを確認してください。また、ジョブが完了したときに実行される done() メソッドもあります。

于 2013-04-13T08:24:08.057 に答える