生产者-消费者（许多）。消费者获取并放入共享队列

Question

我制作了一个生产者-消费者程序。它只是核心 java 中的一个程序，没有任何 GUI（Swing 或 SWT）。它有一个将对象放入队列的生产者。 还有一些消费者必须将一些人员（例如字符串）添加到该共享队列中的每个对象中。因此，每个消费者都必须处理共享队列中的每个对象。 在这种情况下 - 每个 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
    }
}
}

}

这是消费者类之一：

 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();
}
 }

Answer 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
            }
        }
    }

Answer 2

我会尝试改用SwingWorker。它有一个在完成时执行的done() 方法。有关一些代码示例，请参见 this 页面。

如果你使用的不是 Swing，Swt 中有一个类似的功能叫做 Jobs。查看this 页面以获取示例。它还有一个在作业完成时执行的 done() 方法。

Answer 3

还有一些（N 个）消费者必须将一些人员（例如字符串）添加到该共享队列中的每个对象中

我假设您的意思是每个消费者都必须将他们的东西添加到每个进入队列的对象中。在这种情况下，这不是生产者-消费者问题，这更像是一个观察者可观察的问题。基本上，当创建一个新的BookShelf 时，即为Observable。所有Observers 都应该被告知BookShelf，并有机会添加他们自己的Book。

Answer 4

我建议在 Bookshelf 中使用 ConcurrentLinkedQueue 而不是同步列表 - 它是无锁的（不需要同步）并且可能会更有效。

要结束您的消费者，请将他们的 while(true) 循环更改为 while(!cancel) 循环。给每个消费者一个 cancel 布尔值作为初始化为 false 的实例变量，并给他们一个 cancel() 方法，将 cancel 设置为 true。完成后，请致电 cancel() 联系您的消费者。如果您总是一次取消所有消费者（而不是选择性地取消一些但不取消其他消费者），那么您可以使用静态cancel 而不是实例cancel。