JAVA AQS同步队列与条件队列交互导致线程丢失唤醒的分析

JAVA AQS同步队列与条件队列交互导致线程丢失唤醒的分析

大家好，今天我们来深入探讨一个在使用 Java AQS (AbstractQueuedSynchronizer) 时可能会遇到的棘手问题：同步队列与条件队列交互导致的线程丢失唤醒。这个问题在并发编程中比较隐蔽，但理解其原理和解决方案至关重要，能帮助我们编写更健壮的并发程序。

AQS 基础回顾

首先，我们快速回顾一下 AQS 的核心概念：

• 同步状态 (State): AQS 内部维护一个 volatile int 类型的状态变量，用于表示锁的状态或资源可用性。
• 同步队列 (Sync Queue): 也称为 CLH 队列，是一个FIFO双向链表，用于存放等待获取同步状态的线程。当线程竞争同步状态失败时，会被封装成一个 Node 对象，加入到同步队列的尾部。
• 条件队列 (Condition Queue): 每个 Condition 对象内部维护一个条件队列，也是一个FIFO单向链表。当线程调用 Condition.await() 方法时，会被封装成一个 Node 对象，加入到条件队列的尾部，并释放持有的同步状态。当其他线程调用 Condition.signal() 或 signalAll() 方法时，会从条件队列中唤醒一个或所有线程，并将这些线程转移到同步队列中，重新竞争同步状态。

AQS 的核心方法：

方法名	作用
tryAcquire(int)	尝试以独占模式获取同步状态。成功返回 true，失败返回 false。 需要自定义同步器实现。
tryRelease(int)	尝试以独占模式释放同步状态。成功返回 true，失败返回 false。 需要自定义同步器实现。
acquire(int)	以独占模式获取同步状态。如果获取失败，则将当前线程加入同步队列，并阻塞。
release(int)	以独占模式释放同步状态。释放成功后，会唤醒同步队列中第一个等待的线程。
Condition.await()	释放持有的同步状态，并将当前线程加入条件队列，并阻塞。
Condition.signal()	唤醒条件队列中第一个等待的线程，将其转移到同步队列。
Condition.signalAll()	唤醒条件队列中所有等待的线程，将它们转移到同步队列。

线程丢失唤醒的场景分析

现在，我们来看一个经典的线程丢失唤醒的场景。假设我们有一个简单的生产者-消费者模型，使用 AQS 的同步队列和条件队列来实现线程间的同步和通信。

import java.util.concurrent.locks.AbstractQueuedSynchronizer;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

public class FIFOMutex implements Lock {

    private final Sync sync;

    public FIFOMutex() {
        sync = new Sync();
    }

    private static class Sync extends AbstractQueuedSynchronizer {
        @Override
        protected boolean tryAcquire(int acquires) {
            if (compareAndSetState(0, acquires)) {
                setExclusiveOwnerThread(Thread.currentThread());
                return true;
            }
            return false;
        }

        @Override
        protected boolean tryRelease(int releases) {
            if (!isHeldExclusively()) {
                throw new IllegalMonitorStateException();
            }
            if (compareAndSetState(1, 0)) {
                setExclusiveOwnerThread(null);
                return true;
            }
            return false;
        }

        @Override
        protected boolean isHeldExclusively() {
            return getExclusiveOwnerThread() == Thread.currentThread();
        }

        Condition newCondition() {
            return new ConditionObject();
        }
    }

    private Condition newCondition() {
        return sync.newCondition();
    }

    @Override
    public void lock() {
        sync.acquire(1);
    }

    @Override
    public void lockInterruptibly() throws InterruptedException {
        sync.acquireInterruptibly(1);
    }

    @Override
    public boolean tryLock() {
        return sync.tryAcquire(1);
    }

    @Override
    public boolean tryLock(long time, java.util.concurrent.TimeUnit unit) throws InterruptedException {
        return sync.tryAcquireNanos(1, unit.toNanos(time));
    }

    @Override
    public void unlock() {
        sync.release(1);
    }

    @Override
    public Condition newCondition() {
        return sync.newCondition();
    }
}

import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

public class ProducerConsumer {

    private final Lock lock = new FIFOMutex();
    private final Condition notFull  = lock.newCondition();
    private final Condition notEmpty = lock.newCondition();

    private final Queue<Integer> queue = new LinkedList<>();
    private final int capacity;

    public ProducerConsumer(int capacity) {
        this.capacity = capacity;
    }

    public void produce(int item) throws InterruptedException {
        lock.lock();
        try {
            while (queue.size() == capacity) {
                System.out.println("Producer is waiting - queue is full");
                notFull.await();
            }
            queue.offer(item);
            System.out.println("Produced: " + item);
            notEmpty.signal();
        } finally {
            lock.unlock();
        }
    }

    public int consume() throws InterruptedException {
        lock.lock();
        try {
            while (queue.isEmpty()) {
                System.out.println("Consumer is waiting - queue is empty");
                notEmpty.await();
            }
            int item = queue.poll();
            System.out.println("Consumed: " + item);
            notFull.signal();
            return item;
        } finally {
            lock.unlock();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        ProducerConsumer pc = new ProducerConsumer(5);

        Thread producer = new Thread(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    pc.produce(i);
                    Thread.sleep(100);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        Thread consumer = new Thread(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    pc.consume();
                    Thread.sleep(200);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        producer.start();
        consumer.start();

        producer.join();
        consumer.join();

        System.out.println("Producer and Consumer finished.");
    }
}

在这个例子中，ProducerConsumer 类使用 FIFOMutex (一个简单的基于 AQS 的互斥锁) 和两个 Condition 对象 notFull 和 notEmpty 来控制生产者和消费者的行为。

现在，考虑以下场景：

1. 队列已满，生产者线程 P 调用 notFull.await()，释放锁并进入 notFull 条件队列。
2. 消费者线程 C 从队列中取出一个元素，然后调用 notFull.signal() 试图唤醒生产者 P。
3. 关键问题： 在 C 线程调用 notFull.signal() 之后，但在 P 线程被真正唤醒并转移到同步队列之前，另一个线程 C2 (另一个消费者) 抢先获取了锁。
4. C2 发现队列仍然不为空（因为 C 刚刚取出一个元素），于是又从队列中取出一个元素。
5. 现在，C2 释放锁，P 线程被唤醒，并成功获取锁。
6. 丢失唤醒： P 线程认为它被唤醒是因为队列已经不满，但实际上，队列可能仍然是满的！因为 C2 已经消费了一个元素，而 P 还没有生产新的元素。

这种情况下，P 线程可能会错误地认为队列有空间，并尝试生产数据，导致数据溢出或逻辑错误。

原因分析

上述问题的根本原因是 signal() 方法只是将等待线程从条件队列转移到同步队列，并不会立即执行该线程。 在这两个队列之间存在一个时间窗口，其他线程有机会抢先获取锁，并改变条件。

更具体地说，以下是导致线程丢失唤醒的关键步骤：

1. await() 释放锁： Condition.await() 会原子地释放锁，并将当前线程封装成一个 Node 加入到条件队列的尾部，然后阻塞。
2. signal() 转移线程： Condition.signal() 会从条件队列的头部移除一个 Node，并将其加入到同步队列的尾部。这个过程并不会立即执行该 Node 对应的线程。
3. 竞争锁： 从条件队列转移到同步队列的线程需要重新竞争锁。 如果有其他线程（如 C2）在此时获取锁，并改变了条件（例如，消费了队列中的元素），那么被唤醒的线程（如 P）可能会基于过时的条件做出错误的判断。

解决方案

有几种方法可以解决这个问题：

1. 使用 signalAll()：

最简单的解决方案是用 signalAll() 替换 signal()。 signalAll() 会唤醒条件队列中的所有线程，确保至少有一个线程能够满足条件。

    public void produce(int item) throws InterruptedException {
        lock.lock();
        try {
            while (queue.size() == capacity) {
                System.out.println("Producer is waiting - queue is full");
                notFull.await();
            }
            queue.offer(item);
            System.out.println("Produced: " + item);
            notEmpty.signalAll(); // Use signalAll
        } finally {
            lock.unlock();
        }
    }

    public int consume() throws InterruptedException {
        lock.lock();
        try {
            while (queue.isEmpty()) {
                System.out.println("Consumer is waiting - queue is empty");
                notEmpty.await();
            }
            int item = queue.poll();
            System.out.println("Consumed: " + item);
            notFull.signalAll(); // Use signalAll
            return item;
        } finally {
            lock.unlock();
        }
    }

虽然 signalAll() 简单易用，但效率较低。它会唤醒所有等待的线程，即使只有一个线程能够满足条件。 这会导致一些线程被不必要地唤醒，并进行无用的竞争锁操作，从而降低性能。

2. 循环检查条件：

即使使用 signal()，也要在被唤醒后循环检查条件，确保条件仍然满足。 这可以防止线程基于过时的条件做出错误的判断。

    public void produce(int item) throws InterruptedException {
        lock.lock();
        try {
            while (queue.size() == capacity) {
                System.out.println("Producer is waiting - queue is full");
                notFull.await();
            }
            queue.offer(item);
            System.out.println("Produced: " + item);
            notEmpty.signal();
        } finally {
            lock.unlock();
        }
    }

    public int consume() throws InterruptedException {
        lock.lock();
        try {
            while (queue.isEmpty()) {
                System.out.println("Consumer is waiting - queue is empty");
                notEmpty.await();
            }
            int item = queue.poll();
            System.out.println("Consumed: " + item);
            notFull.signal();
            return item;
        } finally {
            lock.unlock();
        }
    }

注意： 尽管上述代码与原始代码相同，但是 while 循环本身就是关键。 即使使用 signal(), await() 总是被包裹在 while 循环中。 线程在被唤醒后，会再次检查条件是否满足。这可以防止 spurious wakeups 和线程丢失唤醒。

3. 使用 ReentrantLock 和 Condition 的公平模式：

ReentrantLock 提供了公平锁的选项。 公平锁会按照线程请求锁的顺序来分配锁，可以减少线程抢占的机会，从而降低线程丢失唤醒的概率。 但是，公平锁会降低并发性能，因为它需要维护一个额外的队列来记录线程的请求顺序。

import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

public class ProducerConsumerFair {

    private final ReentrantLock lock = new ReentrantLock(true); // Fair lock
    private final Condition notFull  = lock.newCondition();
    private final Condition notEmpty = lock.newCondition();

    private final Queue<Integer> queue = new LinkedList<>();
    private final int capacity;

    public ProducerConsumerFair(int capacity) {
        this.capacity = capacity;
    }

    public void produce(int item) throws InterruptedException {
        lock.lock();
        try {
            while (queue.size() == capacity) {
                System.out.println("Producer is waiting - queue is full");
                notFull.await();
            }
            queue.offer(item);
            System.out.println("Produced: " + item);
            notEmpty.signal();
        } finally {
            lock.unlock();
        }
    }

    public int consume() throws InterruptedException {
        lock.lock();
        try {
            while (queue.isEmpty()) {
                System.out.println("Consumer is waiting - queue is empty");
                notEmpty.await();
            }
            int item = queue.poll();
            System.out.println("Consumed: " + item);
            notFull.signal();
            return item;
        } finally {
            lock.unlock();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        ProducerConsumerFair pc = new ProducerConsumerFair(5);

        Thread producer = new Thread(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    pc.produce(i);
                    Thread.sleep(100);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        Thread consumer = new Thread(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    pc.consume();
                    Thread.sleep(200);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        producer.start();
        consumer.start();

        producer.join();
        consumer.join();

        System.out.println("Producer and Consumer finished.");
    }
}

4. 使用更高级的并发工具：

Java 并发包中提供了许多更高级的并发工具，例如 BlockingQueue，Semaphore，CountDownLatch 等。 这些工具已经封装了线程同步和通信的细节，可以避免手动使用 AQS 和条件队列带来的复杂性。 例如，可以使用 BlockingQueue 来实现生产者-消费者模型，而无需手动管理锁和条件变量。

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public class ProducerConsumerBlockingQueue {

    private final BlockingQueue<Integer> queue;

    public ProducerConsumerBlockingQueue(int capacity) {
        this.queue = new LinkedBlockingQueue<>(capacity);
    }

    public void produce(int item) throws InterruptedException {
        System.out.println("Produced: " + item);
        queue.put(item);
    }

    public int consume() throws InterruptedException {
        int item = queue.take();
        System.out.println("Consumed: " + item);
        return item;
    }

    public static void main(String[] args) throws InterruptedException {
        ProducerConsumerBlockingQueue pc = new ProducerConsumerBlockingQueue(5);

        Thread producer = new Thread(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    pc.produce(i);
                    Thread.sleep(100);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        Thread consumer = new Thread(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    pc.consume();
                    Thread.sleep(200);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        producer.start();
        consumer.start();

        producer.join();
        consumer.join();

        System.out.println("Producer and Consumer finished.");
    }
}

总结

线程丢失唤醒是 AQS 同步队列与条件队列交互时可能出现的一个问题。 理解其原因，主要是线程从条件队列转移到同步队列后，需要重新竞争锁，导致条件可能发生变化。 解决方案包括使用 signalAll()，循环检查条件，使用公平锁，以及使用更高级的并发工具。 选择哪种方案取决于具体的应用场景和性能要求。 在编写并发程序时，务必小心处理线程间的同步和通信，避免出现线程丢失唤醒等问题。