生产者/消费者多线程

Question

背景

因为没钱上学，我在收费站上夜班，并利用互联网自学一些编码技能，希望明天能找到更好的工作或在线销售我制作的一些应用程序。漫漫长夜，顾客寥寥。

我正在将多线程作为一个主题来处理，因为我在文献中遇到了很多使用它的代码（例如 Android SDK），但我仍然觉得它晦涩难懂。

精神

此时我的方法是：尝试编写我能想到的最基本的多线程示例，将我的头靠在墙上，看看我是否可以伸展我的大脑容纳一些新颖的思维方式。我将自己暴露在我的极限中，希望能超越它们。随意疯狂批评，甚至吹毛求疵，并指出更好的方法来做我想做的事情。

目标

• Get some advice on how to do the above, based on my efforts so far (code provided)

练习

这是我定义的范围：

定义

创建两个类，它们在数据对象的生产和消费方面协同工作。一个 Thread 创建对象并将它们传送到共享空间，以供另一个获取和使用。我们称生产线程Producer、消费线程Consumer 和共享空间SharedSpace。生产物品供他人消费的行为可以通过类比的方式来同化：

`Producer`    (a busy mum making chocolate-covered cakes for his child, up to a limit)
`Consumer`    (a hungry child waiting to eat all cakes the mum makes, until told to stop)
`SharedSpace` (a kitchen table on which the cakes are put as soon as they become ready)
`dataValue`   (a chocolate-dripping cake which MUST be eaten immediately or else...)

为了简化练习，我决定不让妈妈在孩子吃蛋糕的时候做饭。她只会等待孩子吃完蛋糕，然后立即再做一个，达到一定的限度，以进行良好的育儿。练习的本质是练习Threads 的信号，而不是实现任何并发。相反，我专注于完美的序列化，没有轮询或“我可以走了吗？”检查。我想我将不得不编写后续练习，其中母亲和孩子接下来并行“工作”。

方法

• 让我的类实现 Runnable 接口，以便它们有自己的代码入口点

• 将我的类用作Thread对象的构造函数参数，这些对象从程序的main入口点实例化和启动

• 通过Thread.join()

确保main程序不会在Threads之前终止

• 限制Producer 为Consumer 创建数据的次数

• 就一个 sentinel 值达成一致，Produce 将使用该值表示数据生产结束

• 日志获取共享资源的锁和数据生产/消费事件，包括工作线程的最终注销

• 从程序的main 创建一个SharedSpace 对象，并在启动前将其传递给每个worker

• 在每个工作人员内部存储对SharedSpace对象的private引用

• 提供防护和消息来描述 Consumer 在生成任何数据之前准备好使用的情况

• 在给定的迭代次数后停止Producer

• Consumer 读取哨兵值后停止它

代码

---

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
class Consumer extends Threaded {
  public Consumer(SharedSpace sharedSpace) {
    super(sharedSpace);
  }
  @Override
  public void run() {
    super.run();
    int consumedData = 0;
    while (consumedData != -1) {
      synchronized (sharedSpace) {
        logger.info("Acquired lock on sharedSpace.");
        consumedData = sharedSpace.dataValue;
        if (consumedData == 0) {
          try {
            logger.info("Data production has not started yet. "
                + "Releasing lock on sharedSpace, "
                + "until notification that it has begun.");
            sharedSpace.wait();
          } catch (InterruptedException interruptedException) {
            logger.error(interruptedException.getStackTrace().toString());
          }
        } else if (consumedData == -1) {
          logger.info("Consumed: END (end of data production token).");
        } else {
          logger.info("Consumed: {}.", consumedData);
          logger.info("Waking up producer to continue data production.");
          sharedSpace.notify();
          try {
            logger.info("Releasing lock on sharedSpace "
                + "until notified of new data availability.");
            sharedSpace.wait();
          } catch (InterruptedException interruptedException) {
            logger.error(interruptedException.getStackTrace().toString());
          }
        }
      }
    }
    logger.info("Signing off.");
  }
}
class Producer extends Threaded {
  private static final int N_ITERATIONS = 10;
  public Producer(SharedSpace sharedSpace) {
    super(sharedSpace);
  }
  @Override
  public void run() {
    super.run();
    int nIterations = 0;
    while (nIterations <= N_ITERATIONS) {
      synchronized (sharedSpace) {
        logger.info("Acquired lock on sharedSpace.");
        nIterations++;
        if (nIterations <= N_ITERATIONS) {
          sharedSpace.dataValue = nIterations;
          logger.info("Produced: {}", nIterations);
        } else {
          sharedSpace.dataValue = -1;
          logger.info("Produced: END (end of data production token).");
        }
        logger.info("Waking up consumer for data consumption.");
        sharedSpace.notify();
        if (nIterations <= N_ITERATIONS) {
          try {
            logger.info("Releasing lock on sharedSpace until notified.");
            sharedSpace.wait();
          } catch (InterruptedException interruptedException) {
            logger.error(interruptedException.getStackTrace().toString());
          }
        }
      }
    }
    logger.info("Signing off.");
  }
}
class SharedSpace {
  volatile int dataValue = 0;
}
abstract class Threaded implements Runnable {
  protected Logger logger;
  protected SharedSpace sharedSpace;
  public Threaded(SharedSpace sharedSpace) {
    this.sharedSpace = sharedSpace;
    logger = LoggerFactory.getLogger(this.getClass());
  }
  @Override
  public void run() {
    logger.info("Started.");
    String workerName = getClass().getName();
    Thread.currentThread().setName(workerName);
  }
}
public class ProducerConsumer {
  public static void main(String[] args) {
    SharedSpace sharedSpace = new SharedSpace();
    Thread producer = new Thread(new Producer(sharedSpace), "Producer");
    Thread consumer = new Thread(new Consumer(sharedSpace), "Consumer");
    producer.start();
    consumer.start();
    try {
      producer.join();
      consumer.join();
    } catch (InterruptedException interruptedException) {
      interruptedException.printStackTrace();
    }
  }
}

---

执行日志

---

Consumer - Started.
Consumer - Acquired lock on sharedSpace.
Consumer - Data production has not started yet. Releasing lock on sharedSpace, until notification that it has begun.
Producer - Started.
Producer - Acquired lock on sharedSpace.
Producer - Produced: 1
Producer - Waking up consumer for data consumption.
Producer - Releasing lock on sharedSpace until notified.
Consumer - Acquired lock on sharedSpace.
Consumer - Consumed: 1.
Consumer - Waking up producer to continue data production.
Consumer - Releasing lock on sharedSpace until notified of new data availability.
Producer - Acquired lock on sharedSpace.
Producer - Produced: 2
Producer - Waking up consumer for data consumption.
Producer - Releasing lock on sharedSpace until notified.
Consumer - Acquired lock on sharedSpace.
Consumer - Consumed: 2.
Consumer - Waking up producer to continue data production.
Consumer - Releasing lock on sharedSpace until notified of new data availability.
Producer - Acquired lock on sharedSpace.
Producer - Produced: 3
Producer - Waking up consumer for data consumption.
Producer - Releasing lock on sharedSpace until notified.
Consumer - Acquired lock on sharedSpace.
Consumer - Consumed: 3.
Consumer - Waking up producer to continue data production.
Consumer - Releasing lock on sharedSpace until notified of new data availability.
Producer - Acquired lock on sharedSpace.
Producer - Produced: 4
Producer - Waking up consumer for data consumption.
Producer - Releasing lock on sharedSpace until notified.
Consumer - Acquired lock on sharedSpace.
Consumer - Consumed: 4.
Consumer - Waking up producer to continue data production.
Consumer - Releasing lock on sharedSpace until notified of new data availability.
Producer - Acquired lock on sharedSpace.
Producer - Produced: 5
Producer - Waking up consumer for data consumption.
Producer - Releasing lock on sharedSpace until notified.
Consumer - Acquired lock on sharedSpace.
Consumer - Consumed: 5.
Consumer - Waking up producer to continue data production.
Consumer - Releasing lock on sharedSpace until notified of new data availability.
Producer - Acquired lock on sharedSpace.
Producer - Produced: 6
Producer - Waking up consumer for data consumption.
Producer - Releasing lock on sharedSpace until notified.
Consumer - Acquired lock on sharedSpace.
Consumer - Consumed: 6.
Consumer - Waking up producer to continue data production.
Consumer - Releasing lock on sharedSpace until notified of new data availability.
Producer - Acquired lock on sharedSpace.
Producer - Produced: 7
Producer - Waking up consumer for data consumption.
Producer - Releasing lock on sharedSpace until notified.
Consumer - Acquired lock on sharedSpace.
Consumer - Consumed: 7.
Consumer - Waking up producer to continue data production.
Consumer - Releasing lock on sharedSpace until notified of new data availability.
Producer - Acquired lock on sharedSpace.
Producer - Produced: 8
Producer - Waking up consumer for data consumption.
Producer - Releasing lock on sharedSpace until notified.
Consumer - Acquired lock on sharedSpace.
Consumer - Consumed: 8.
Consumer - Waking up producer to continue data production.
Consumer - Releasing lock on sharedSpace until notified of new data availability.
Producer - Acquired lock on sharedSpace.
Producer - Produced: 9
Producer - Waking up consumer for data consumption.
Producer - Releasing lock on sharedSpace until notified.
Consumer - Acquired lock on sharedSpace.
Consumer - Consumed: 9.
Consumer - Waking up producer to continue data production.
Consumer - Releasing lock on sharedSpace until notified of new data availability.
Producer - Acquired lock on sharedSpace.
Producer - Produced: 10
Producer - Waking up consumer for data consumption.
Producer - Releasing lock on sharedSpace until notified.
Consumer - Acquired lock on sharedSpace.
Consumer - Consumed: 10.
Consumer - Waking up producer to continue data production.
Consumer - Releasing lock on sharedSpace until notified of new data availability.
Producer - Acquired lock on sharedSpace.
Producer - Produced: END (end of data production token).
Producer - Waking up consumer for data consumption.
Producer - Signing off.
Consumer - Acquired lock on sharedSpace.
Consumer - Consumed: END (end of data production token).
Consumer - Signing off.

---

问题

• 以上是否正确？ （例如，它是否使用了正确的语言工具、正确的方法、是否包含任何愚蠢的代码……）

但它“看起来不错”？

即使输出“看起来不错”，我也会询问正确性，因为您无法想象在我的测试“一次”而不是“另一次”中出现了多少次错误（例如，当消费者首先开始时，当生产者在生产哨兵后永远不会退出等）。我学会了不要从“成功的运行”中声称正确。相反，我对伪并行代码变得非常怀疑！ （根据定义，这个甚至不是平行的！0

扩展答案

一个很好的问题只关注one requested piece of advice（上述问题），但如果您愿意，请随时在您的回答中提及对以下其他主题的任何见解：

• 在编写下一次尝试时如何测试并行代码？

• 哪些工具可以帮助我进行开发和调试？考虑我使用 Eclipse

• 如果我允许Producer 继续生产，每次生产需要一些可变的时间，而Consumer 消耗任何可用的东西，方法会改变吗？锁定是否必须移动到其他地方？信号是否需要改变这种等待/通知范式？

• 这种做事方法已过时吗？我应该学习其他东西吗？从这个收费站，我不知道“在 Java 的真实世界中”会发生什么

后续步骤

• 我应该从这里去哪里？我曾在某处看到过“期货”的概念，但我可以使用主题编号列表按教学顺序进行操作，并附上相关学习资源的链接

---

蒂诺西诺

Answer 1

上面说的对吗？

我看到的唯一问题是@Tudor 和@Bhaskar 提到的问题。每当您在等待某个条件时对其进行测试，您必须使用while 循环。然而，这更多的是与多个生产者和消费者的竞争条件。可能会发生虚假唤醒，但竞争条件更有可能发生。见my page on the subject。

是的，您只有 1 个生产者和 1 个消费者，但您可以尝试为多个消费者扩展您的代码或将您的代码复制到另一个场景。

我学会了不要从“成功的运行”中声称正确。反而对伪并行代码产生了很大的怀疑！

良好的直觉。

我如何在下次尝试编写代码时测试并行代码？

这很难。扩大规模是一种方法。添加多个生产者和消费者，看看是否有问题。在具有不同数量/类型的处理器的多个架构上运行。你最好的防御将是代码的正确性。紧密同步，善用BlockingQueue、ExecutorService等类，让你的关闭更简单/更干净。

没有简单的答案。测试多线程代码非常困难。

哪些工具可以帮助我进行开发和调试？

就一般性而言，我会研究像Emma 这样的覆盖工具，这样您就可以确保您的单元测试覆盖了您的所有代码。

在多线程代码测试方面，了解如何阅读kill -QUIT thread-dumps 并查看 Jconsole 内部正在运行的线程。像 YourKit 这样的 Java 分析器也可能会有所帮助。

如果我允许 Producer 继续制作，每次制作都需要不同的时间，方法会改变吗...

我不这么认为。消费者将永远等待生产者。可能我没看懂这个问题？

这种做事方法是否已经过时，我应该学习其他东西吗？从这个收费站，我不知道“在 Java 的真实世界中”会发生什么

接下来是了解ExecutorService classes。它们处理大部分new Thread() 风格的代码——尤其是当您处理大量使用线程执行的异步任务时。这是tutorial。

我应该从这里去哪里？

再次，ExecutorService。我假设你已经阅读了this starting docs。正如@Bhaskar 提到的，Java Concurrency in Practice 是一本好圣经。

---

以下是关于您的代码的一些通用 cmets：

• SharedSpace 和 Threaded 类似乎是一种人为的方式来做到这一点。如果您正在玩基类等，那很好。但总的来说，我从不使用这样的模式。生产者和消费者通常使用BlockingQueue，例如LinkedBlockingQueue，在这种情况下，同步和volatile 有效负载会为您处理好。另外，我倾向于将共享信息注入对象构造函数，而不是从基类中获取。

• 通常，如果我使用synchronized，它位于private final 字段上。通常我会创建一个 private final Object lockObject = new Object(); 用于锁定，除非我已经在使用一个对象。

• 注意巨大的synchronized 块并将日志消息放在synchronized 部分内。日志通常对文件系统执行synchronized IO，这可能非常昂贵。如果可能的话，你应该有小的、非常紧的 synchronized 块。

• 您在循环之外定义consumedData。我会在分配时定义它，然后使用break 来从循环中退出（如果它是== -1）。如果可能，请确保限制您的局部变量范围。

• 您的日志消息将主导您的代码性能。这意味着当您删除它们时，您的代码将完全以不同的方式执行。当您使用它来调试问题时，意识到这一点非常很重要。当您迁移到具有不同 CPU/内核的不同架构时，性能也（很可能）会发生变化。

• 您可能知道这一点，但是当您调用sharedSpace.notify(); 时，这仅意味着通知另一个线程如果它当前位于sharedSpace.wait();。如果它不是其他东西，那么它将错过通知。仅供参考。

• 做一个if (nIterations <= N_ITERATIONS)有点奇怪，然后else下面的3行再做一次。复制notify() 会更好地简化分支。

• 你有一个int nIterations = 0;，然后是一个while，然后是一个++。这是 for 循环的秘诀：

  for (int nIterations = 0; nIterations <= N_ITERATIONS; nIterations++) {
  

---

这是您的代码的更紧凑的版本。这只是我如何编写它的一个例子。同样，除了缺少while 之外，您的版本似乎没有任何问题。

public class Consumer implements Runnable {
    private final BlockingQueue<Integer> queue;
    public Consumer(BlockingQueue<Integer> queue) {
       this.queue = queue;
    }
    @Override
    public void run() {
       while (true) {
          int consumedData = queue.take();
          if (consumedData ==  Producer.FINAL_VALUE) {
              logger.info("Consumed: END (end of data production token).");
              break;
          }
          logger.info("Consumed: {}.", consumedData);
       }
       logger.info("Signing off.");
    }
}

public class Producer implements Runnable {
    public static final int FINAL_VALUE = -1;
    private final BlockingQueue<Integer> queue;
    public Producer(BlockingQueue<Integer> queue) {
       this.queue = queue;
    }
    @Override
    public void run() {
       for (int nIterations = 0; nIterations <= N_ITERATIONS; nIterations++) {
          logger.info("Produced: {}", nIterations);
          queue.put(nIterations);
       }
       queue.put(FINAL_VALUE);
       logger.info("Produced: END (end of data production token).");
       logger.info("Signing off.");
    }
}

public class ProducerConsumer {
    public static void main(String[] args) {
       // you can add an int argument to the LinkedBlockingQueue constructor
       // to only allow a certain number of items in the queue at one time
       BlockingQueue<Integer> queue = new LinkedBlockingQueue<Integer>();
       Thread producer = new Thread(new Producer(queue), "Producer");
       Thread consumer = new Thread(new Consumer(queue), "Consumer");
       // start and join go here
    }
}

Answer 2

你似乎在这里做得很好。其实没什么好挑剔的。我想推荐的一个想法是你应该避免在缓冲区对象本身上进行同步。在这种情况下没关系，但是假设您切换到数据结构缓冲区，这取决于它可能在内部同步的类（例如Vector，尽管它现在已经过时了），所以从外部获取锁可能会搞砸.

编辑：Bhaskar 提出了一个很好的观点，即使用while 来包装对wait 的调用。这是因为可能会发生臭名昭著的虚假唤醒，迫使线程过早退出wait，因此您需要确保它重新进入。

接下来你可以做的是实现一个有限缓冲区的生产者消费者：拥有一些共享的数据结构，例如一个链表并设置最大大小（例如 10 个项目）。然后让生产者继续生产，并且只有在队列中有 10 个项目时才暂停它。只要缓冲区为空，消费者就会被挂起。

接下来您可以采取的步骤是学习如何将您手动实施的流程自动化。看看BlockingQueue，它提供了一个具有阻塞行为的缓冲区（即，如果缓冲区为空，消费者将自动阻塞，如果缓冲区已满，生产者将阻塞）。

此外，根据具体情况，executors（查看ExecutorService）可能是一个值得替代的替代品，因为它们封装了一个任务队列和一个或多个工人（消费者），所以你只需要生产者。

Answer 3

Producers 和Consumers 可以是实现Runnable 的简单类（没有extends Threaded）这样它们就不那么脆弱了。客户端可以自己创建Threads 并附加实例，因此不需要类层次结构的开销。

wait() 之前的条件应该是 while() 而不是 if 。

编辑：来自 JCIP 第 301 页：

void stateDependentMethod() throws InterruptedException {
      // condition predicate must be guarded by lock
      synchronized(lock) {
          while (!conditionPredicate())
            lock.wait();
          // object is now in desired state
       }
  }

您已经内置了静态停止的条件。通常，生产者和消费者应该更加灵活——他们应该能够响应外部信号停止。

对于初学者来说，要实现外部停止信号，你有一个标志：

class Producer implements Runnable { 
     private volatile boolean stopRequested ;

     public void run() {
        while(true){
           if(stopRequested )
                // get out of the loop
         }
     }

     public void stop(){
        stopRequested  = true;
        // arrange to  interrupt the Producer thread here.
     }
 }

当您尝试执行上述操作时，您可能会看到出现其他复杂情况 - 例如 - 您的制作人首先发布然后 wait() ing，但这可能会导致问题。

如果您有兴趣进一步阅读，我建议您阅读本书-Java Concurrency In Practice。这将有很多建议，我无法在此处添加。

Answer 4

雄心壮志！大约 8 年前，你问过这个问题。我希望您的努力为您提供（并继续为您提供）您想要的教育。

如今，强烈建议不要使用 wait()、notify() 和 join() 在 Java 中实现多线程。当你试图在这个低级别控制并发时，太容易自取其辱了（事实上，Java 设计者承认 Thread 的许多方法和语义实际上是设计错误，但他们不得不将它们留在后面出于兼容性目的——许多人正在使用新的“虚拟线程”（Project Loom）——但这是一个不同的话题。

今天手动启动和控制线程的首选方式是通过ExecutorService.submit(Callable<V>)，返回Future<V>。然后，您可以通过调用Future<V>.get() 来等待线程退出（并获取返回值），返回由可调用对象返回的V 类型的值（如果Callable 抛出未捕获的对象，则抛出ExecutionException例外）。

下面的类是一个如何实现类似的例子。这将通过单个有界阻塞队列将任意数量的生产者连接到任意数量的消费者。 （来自线程的返回值被忽略，因此调用ExecutorService.submit(Runnable)，返回Future<?>，而不是ExecutorService.submit(Callable<V>)）。

import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public abstract class ProducerConsumer<E> {

    private final BlockingQueue<Optional<E>> queue;

    public ProducerConsumer(
            int numProducerThreads, int numConsumerThreads, int queueCapacity) {
        if (numProducerThreads < 1 || numConsumerThreads < 1 || queueCapacity < 1) {
            throw new IllegalArgumentException();
        }
        queue = new ArrayBlockingQueue<Optional<E>>(queueCapacity);
        final ExecutorService executor = 
                Executors.newFixedThreadPool(numProducerThreads + numConsumerThreads);
        try {
            // Start producer threads
            final List<Future<?>> producerFutures = new ArrayList<>();
            final AtomicInteger numLiveProducers = new AtomicInteger();
            for (int i = 0; i < numProducerThreads; i++) {
                producerFutures.add(executor.submit(() -> {
                    numLiveProducers.incrementAndGet();
                    // Run producer
                    producer();
                    // When last producer finishes, deliver poison pills to consumers
                    if (numLiveProducers.decrementAndGet() == 0) {
                        for (int j = 0; j < numConsumerThreads; j++) {
                            queue.put(Optional.empty());
                        }
                    }
                    return null;
                }));
            }
            // Start consumer threads
            final List<Future<?>> consumerFutures = new ArrayList<>();
            for (int i = 0; i < numConsumerThreads; i++) {
                consumerFutures.add(executor.submit(() -> {
                    // Run Consumer
                    consumer();
                    return null;
                }));
            }
            // Wait for all producers to complete
            completionBarrier(producerFutures, false);
            // Shut down any consumers that are still running after producers complete
            completionBarrier(consumerFutures, false);
        } finally {
            executor.shutdownNow();
        }
    }

    private static void completionBarrier(List<Future<?>> futures, boolean cancel) {
        for (Future<?> future : futures) {
            try {
                if (cancel) {
                    future.cancel(true);
                }
                future.get();
            } catch (CancellationException | InterruptedException e) {
                // Ignore
            } catch (ExecutionException e) {
                throw new RuntimeException(e);
            }
        }
    }

    protected void produce(E val) {
        try {
            queue.put(Optional.of(val));
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    protected Optional<E> consume() {
        try {
            return queue.take();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    /** Producer loop. Call {@link #produce(E)} for each element. */
    public abstract void producer();

    /**
     * Consumer thread. Call {@link #consume()} to get each successive element,
     * until an empty {@link Optional} is returned.
     */
    public abstract void consumer();
}

如下使用：

new ProducerConsumer<Integer>(/* numProducerThreads = */ 1, /* numConsumerThreads = */ 4,
        /* queueCapacity = */ 10) {
    @Override
    public void producer() {
        for (int i = 0; i < 100; i++) {
            System.out.println("Producing " + i);
            produce(i);
        }
    }

    @Override
    public void consumer() {
        for (Optional<Integer> opt; (opt = consume()).isPresent; ) {
            int i = opt.get();
            System.out.println("Got " + i);
        }
    }
};