如何使用容量为1的BlockingQueue解决生产者消费者问题

你遇到的日志顺序异常，本质不是队列逻辑有问题，而是日志打印的时机和线程调度的特性导致的错觉，核心问题有两个：

• 生产者先打印生产日志，再执行put操作：打印日志的时候元素还没有真正入队，甚至put可能接下来就进入阻塞状态，相当于你提前宣告了生产完成，实际生产动作还没执行完毕。
• 队列操作和打印操作没有原子性：就算调整了打印时机，CPU的线程调度是随机的，队列操作完成后当前线程可能没抢到时间片执行打印，另一个线程先执行完了操作和打印，也会导致日志顺序和预期不符。

不同队列的异常日志具体成因 #

1. ArrayBlockingQueue/LinkedBlockingQueue（容量为1）场景
   第一个元素put成功后队列占满，生产者进入下一轮循环，会先打印第二个元素的生产日志，再执行put操作，这时候才会因为队列已满进入阻塞，自然就会出现连续两条生产日志，之后才会打印消费日志。
2. SynchronousQueue场景
   SynchronousQueue本身没有存储容量，put操作必须等待消费者的take操作配对才会返回。但你在put前打印日志，可能出现生产者打印了生产日志后执行put阻塞，还没等消费者打印消费日志，CPU时间片又切回生产者，下一轮循环又打印了新的生产日志再进入阻塞，最终呈现连续生产的日志。

如果要严格保证生产N→消费N的日志顺序，只需调整打印时机，配合SynchronousQueue即可达到效果：

1. 生产者的生产日志挪到put操作成功之后，代表元素真正完成入队才算生产完成。
2. 消费者的打印逻辑无需调整，本身就是take到元素之后才打印。

修复后的代码如下：

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.SynchronousQueue;

class Producent extends Thread {
    private final BlockingQueue<Integer> blockingQueue;

    public Producent(BlockingQueue<Integer> blockingQueue) {
        this.blockingQueue = blockingQueue;
    }

    @Override
    public void run() {
        for (int i = 0; i < 100; ++i) {
            try {
                // 先执行put操作，成功后再打印生产日志
                blockingQueue.put(i);
                System.out.println("PRODUCE: " + i);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

class Konsument extends Thread {
    private final BlockingQueue<Integer> blockingQueue;

    public Konsument(BlockingQueue<Integer> blockingQueue) {
        this.blockingQueue = blockingQueue;
    }

    public void run() {
        for (int i = 0; i < 100; ++i) {
            try {
                consume(blockingQueue.take());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    private void consume(int number) {
        System.out.println("CONSUME: " + number);
    }
}

public class PKmon {
    public static void main(String[] args) {
        BlockingQueue<Integer> blockingQueue = new SynchronousQueue<>();
        Thread producent = new Producent(blockingQueue);
        Thread konsument = new Konsument(blockingQueue);
        producent.start();
        konsument.start();
    }
}

补充说明 #

如果调整后偶尔还是出现顺序偏差，是因为SynchronousQueue的put和take是配对同时返回的，两个线程的打印顺序由CPU调度决定，属于正常现象。如果必须严格保证生产日志永远在对应消费日志前面，可以额外加一个二元信号量控制打印顺序，一般业务场景下只要逻辑上是生产一个消费一个，无需纠结打印的绝对顺序。

内容的提问来源于stack exchange，提问作者MrFisherman