java - 多个消费者线程总体消费队列 FIFO

Question

当我尝试学习JAVA编程的多线程部分时，在处理“一个生产者 - 多个消费者”编码时遇到以下问题。

我想要实现的是:多个消费者线程按照项目放入队列的顺序将项目从队列中取出。也就是说，让消费者线程整体保持先进先出的方式。

final BlockingDeque<String> deque = new LinkedBlockingDeque<String>();

Runnable rb = new Runnable() {
    public void run() {
        try {
            System.out.println(deque.takeLast());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
};

deque.putFirst("a");
deque.putFirst("b");
deque.putFirst("c");
deque.putFirst("d");

ExecutorService pool = Executors.newFixedThreadPool(4);
pool.submit(rb);
pool.submit(rb);
pool.submit(rb);
pool.submit(rb);

我在寻找什么:A乙Cd

它实际输出什么:乙CAd

随机顺序或

有什么简单的解决方案可以解决这个问题吗？谢谢!

Answer 1

就您而言，问题在于

System.out.println(deque.takeLast());

实际上是两条指令，它们一起不是原子的。想象一下这样的场景:

1. 线程 1 从队列中获取字符串。
2. 线程 2 从队列中获取字符串。
3. 线程 2 打印值。
4. 线程 1 打印值。

所以这完全取决于操作系统如何管理线程的执行。

在您的情况下，一种可能的解决方案是将 synchronized 关键字添加到 run 方法:

Runnable rb = new Runnable() {
    public synchronized void run() {
         try {
              String s = deque.takeLast();
              System.out.println(s);
         } catch (InterruptedException e) {
              e.printStackTrace();
         }
    }
};

这将同步您在此处创建的匿名类的实例。由于您将相同的可运行对象传递给 ExecutorService - 它应该可以工作。或者您可以同步您的 queue 对象，因为您的可运行对象可以访问队列对象，当您将其传递给 ExecutorService 时，它将在许多线程中执行:

Runnable rb = new Runnable() {
    public void run() {
        synchronized (deque) {
             try {
                 String s = deque.takeLast();
                 System.out.println(s);
             } catch (InterruptedException e) {
                 e.printStackTrace();
             }
        }
    }
};

还要记住关闭线程池，因为现在您的应用程序将永远不会退出。