FizzBu​​zz 多线程消费者生产者

Question

我有几个线程（生产者和消费者）的 FizzBu​​zz 问题 我在网上找到了这个解决方案：https://gist.github.com/masudak/5098917 但我担心一个特定的情况，一个糟糕的上下文切换，其中'isStop' 是的，但并非所有生产者都完成了将他们的号码添加到队列中。 这是一个真正的问题吗？还是我错过了什么？

public static AtomicInteger counter = new AtomicInteger(0);
public static ConcurrentLinkedQueue<Integer> queue = new ConcurrentLinkedQueue<Integer>();
public static volatile boolean isStop = false; 

public static class Producer implements Runnable{

    @Override
    public void run() {
        while ( true ) {
            Integer counterValue = counter.incrementAndGet();
            if (counterValue <= 100) {
                queue.add(counterValue);
            } else {
                isStop = true;
                break;
            }
        }
    }   
}

public static class Consumer implements Runnable{

    @Override
    public void run() {
        while ( true ) {
            Integer counterValue = queue.poll();
            if ( counterValue == null ){
                if(isStop){
                    break;
                }
                continue;
            }

            fizzBuzz(counterValue);
        }
    }


    private void fizzBuzz(Integer value){
        if( value % 15 == 0 ){
            System.out.println("FizzBuzz:" + value);
        } else if( value % 3 == 0 ){
            System.out.println("Fizz:" + value);
        } else if( value % 5 == 0 ){                
            System.out.println("Buzz:" + value );
        } else {
            System.out.println(value);
        }
    }   
}

Answer 1

尽管格式错误和其他问题，恕我直言，这实际上是一个有趣的问题。是的，我相信这段代码包含一个理论上的问题，尽管这个问题很难暴露。但是，如果您通过添加一些显式随机 Thread.sleep 来稍微修改代码，以模拟现实世界任务中不同线程性能的意外差异，您可以很容易地看到这个问题。然后要重现该问题，您只需要 3 个线程：2 个 Producers 和一个 Consumers。同样重要的是，在添加 sleep 之后，Producers 的速度变得比单个 Consumer 的速度要慢得多（不像原始代码中 Consumers 被 I/O 操作阻塞，而 Producers不是）。我还将最大值减少到 10 以缩短日志。所以这里是代码：

import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicInteger;


public class fizzbuzz {
    public static void main(String[] args) {
        Thread th1 = new Thread(new Producer());
        Thread th2 = new Thread(new Producer());

        Thread th4 = new Thread(new Consumer());

        th1.start();
        th2.start();
        th4.start();
    }

    public static AtomicInteger counter = new AtomicInteger(0);
    public static ConcurrentLinkedQueue<Integer> queue = new ConcurrentLinkedQueue<Integer>();
    public static volatile boolean isStop = false;

    public static class Producer implements Runnable {

        @Override
        public void run() {
            while (true) {
                Integer counterValue = counter.incrementAndGet();
                final int MAX = 10;
                if (counterValue <= MAX) {
                    final int sleep = ThreadLocalRandom.current().nextInt(200);
                    try {
                        Thread.sleep(sleep);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }

                    queue.add(counterValue);
                } else {
                    isStop = true;
                    break;
                }
            }
        }
    }

    public static class Consumer implements Runnable {

        @Override
        public void run() {
            while (true) {
                Integer counterValue = queue.poll();
                if (counterValue == null) {
                    if (isStop) {
                        break;
                    }

                    continue;
                }

                fizzBuzz(counterValue);
            }
        }


        private void fizzBuzz(Integer value) {
            if (value % 15 == 0) {
                System.out.println("FizzBuzz:" + value);
            } else if (value % 3 == 0) {
                System.out.println("Fizz:" + value);
            } else if (value % 5 == 0) {
                System.out.println("Buzz:" + value);
            } else {
                System.out.println(value);
            }
        }
    }

}

如果您多次运行此代码，您可能会看到如下内容：

1
2
4
嘶嘶声：3
嘶嘶声：6
嗡嗡声：5
7
8
嗡嗡声：10

请注意，缺少 9。发生这种情况是因为生产者 #9 的睡眠比生产者 #10 的睡眠要多，所以 isStop 是在实际添加所有元素之前引发的。

甚至是这样的：

1
2
嘶嘶声：3
嗡嗡声：5
4
嘶嘶声：6
7
嘶嘶声：9
8

注意 10 是如何丢失的。恕我直言，这更有趣：生产者 #9 向队列添加了值，并且在生产者 #10 处于休眠状态的下一次迭代中，由于没有更多的东西要生产，因此引发了 isStop 标志。

此类问题的一个明显解决方法是使用CountDownLatch 之类的东西来设置isStop，而不是在第一个制作人完成工作时而是在最后一个制作人完成工作时。