Java实现单生产者多消费者时仅单个消费者消费Buffer的问题排查

DevOps 小满

2026-10-24

核心问题根源

你的代码问题和Buffer实现本身关系不大，核心问题出在锁的持有范围完全失控，具体错误点如下：

外层同步块粒度过大
你将Producer和Consumer的整个无限循环for(;;)都包裹在synchronized(_buf)同步块内部，意味着一旦某个线程抢到_buf对象的锁，只要没有主动调用wait()释放锁，就会一直持有这把锁，其他所有需要竞争这把锁的线程，根本没机会进入自己的同步逻辑。
持锁状态下执行阻塞操作
你把Thread.sleep()调用也写在了同步块内部。Java的线程进入sleep()状态时不会主动释放持有的锁，等于线程拿着锁睡觉，其他所有线程都要等它睡醒、执行完循环逻辑才有机会抢锁。
锁的获取形成排他垄断
按照你代码的执行逻辑：某个消费者线程（比如输出里的CONSUMER 6）一旦在启动阶段抢到_buf的锁，就会进入自己的无限循环：
1. 只要Buffer非空，就连续调用get()消费数据，因为本身持有锁，这个过程完全没有阻塞
2. 消费完成后调用sleep(1000)，全程不释放锁
3. 唤醒后调用notifyAll()，立刻进入下一轮循环，直到Buffer被消费到空，才会调用wait()释放锁
  这期间其他5个消费者从启动开始，就因为拿不到_buf的锁，一直卡在同步块的入口处，根本执行不到判断Buffer状态、消费数据的逻辑。等Buffer空了、持有锁的消费者释放锁进入wait状态，生产者抢到锁开始生产数据、调用notifyAll()唤醒所有等待线程时，之前持有锁的那个消费者会因为锁等待队列的调度优先级，第一时间重新抢到锁，又开始连续垄断消费，永远轮不到其他消费者。
锁的获取逻辑冗余
你在Buffer类的put()、get()方法上已经加了synchronized保证操作安全，外层又在run方法上加了同一把锁，属于重复嵌套加锁，进一步放大了锁的影响范围。

修复方案

按照以下步骤调整代码就能解决单消费者垄断的问题：

移除run方法外层的大同步块，不要把整个生产/消费循环包裹在synchronized(_buf)中，锁的持有范围只需要覆盖Buffer的读写操作即可。
把等待/通知逻辑下沉到Buffer内部：Buffer类自己负责判断满/空状态，在put()方法内部实现满则等待的逻辑，在get()方法内部实现空则等待的逻辑，去掉原来检测到满/空就直接退出程序的逻辑。
把sleep操作移到锁的外面：线程执行sleep模拟消费/生产耗时的时候，不要持有Buffer的锁，让其他线程有机会竞争锁。

调整后核心代码示例：

// 调整后的Buffer类核心方法
class Buffer {
    // 原有队列属性保持不变
    public synchronized void put(int i) throws InterruptedException {
        // 满则等待，替代原来直接退出程序的逻辑
        while (isFull()) {
            System.out.println("Buffer is full, producer waiting");
            wait();
        }
        System.out.println("Inserting " + i);
        rear = (rear + 1) % capacity;
        arr[rear] = i;
        count++;
        notifyAll();
    }

    public synchronized int get() throws InterruptedException {
        // 空则等待，替代原来直接退出程序的逻辑
        while (isEmpty()) {
            System.out.println("Buffer is empty, consumer waiting");
            wait();
        }
        int result = arr[front];
        System.out.println("Removing " + arr[front]);
        front = (front + 1) % capacity;
        count--;
        notifyAll();
        return result;
    }
}

// 调整后的Producer run方法
@Override
public void run() {
    for (; ; ) {
        int produceVal = new Random().nextInt();
        try {
            _buf.put(produceVal);
            System.out.println(this.name + " producing value " + produceVal);
            // sleep在锁外面执行，不持有锁
            Thread.sleep(200);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

// 调整后的Consumer run方法
@Override
public void run() {
    for (; ; ) {
        try {
            int consumeVal = _buf.get();
            System.out.println(this.name + ": consuming value " + consumeVal);
            // sleep在锁外面执行，不持有锁
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

调整完成后，所有消费者会公平竞争Buffer的锁，消费、生产操作会在多个线程间均匀分配，不会再出现单个消费者垄断全部消费任务的问题。

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