Condition案例

本文目录导读：

1. 完整案例：阻塞队列实现
2. 高级案例：多条件协作
3. 公平锁和超时等待示例

我将为您提供一个完整的Java并发编程中Condition的案例,展示生产者-消费者模式。

完整案例：阻塞队列实现

import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
/**
 * 使用Condition实现阻塞队列
 */
public class ConditionExample {
    // 自定义阻塞队列
    static class BlockingQueue<T> {
        private final Queue<T> queue = new LinkedList<>();
        private final int capacity;
        private final ReentrantLock lock = new ReentrantLock();
        // 队列非满条件
        private final Condition notFull = lock.newCondition();
        // 队列非空条件
        private final Condition notEmpty = lock.newCondition();
        public BlockingQueue(int capacity) {
            this.capacity = capacity;
        }
        /**
         * 向队列中添加元素
         */
        public void put(T element) throws InterruptedException {
            lock.lock();
            try {
                // 队列已满时等待
                while (queue.size() == capacity) {
                    System.out.println(Thread.currentThread().getName() + 
                        " 队列已满，等待消费...");
                    notFull.await();
                }
                queue.add(element);
                System.out.println(Thread.currentThread().getName() + 
                    " 生产元素: " + element + ", 队列大小: " + queue.size());
                // 通知消费者可以消费了
                notEmpty.signal();
            } finally {
                lock.unlock();
            }
        }
        /**
         * 从队列中取出元素
         */
        public T take() throws InterruptedException {
            lock.lock();
            try {
                // 队列为空时等待
                while (queue.isEmpty()) {
                    System.out.println(Thread.currentThread().getName() + 
                        " 队列为空，等待生产...");
                    notEmpty.await();
                }
                T element = queue.poll();
                System.out.println(Thread.currentThread().getName() + 
                    " 消费元素: " + element + ", 队列大小: " + queue.size());
                // 通知生产者可以生产了
                notFull.signal();
                return element;
            } finally {
                lock.unlock();
            }
        }
        /**
         * 获取队列大小
         */
        public int size() {
            lock.lock();
            try {
                return queue.size();
            } finally {
                lock.unlock();
            }
        }
    }
    // 生产者
    static class Producer implements Runnable {
        private final BlockingQueue<String> queue;
        private final String name;
        private int count = 0;
        public Producer(BlockingQueue<String> queue, String name) {
            this.queue = queue;
            this.name = name;
        }
        @Override
        public void run() {
            Thread.currentThread().setName(name);
            try {
                while (true) {
                    String element = "数据-" + (++count);
                    queue.put(element);
                    Thread.sleep(500); // 模拟生产耗时
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }
    // 消费者
    static class Consumer implements Runnable {
        private final BlockingQueue<String> queue;
        private final String name;
        public Consumer(BlockingQueue<String> queue, String name) {
            this.queue = queue;
            this.name = name;
        }
        @Override
        public void run() {
            Thread.currentThread().setName(name);
            try {
                while (true) {
                    String element = queue.take();
                    Thread.sleep(1000); // 模拟消费耗时
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }
    public static void main(String[] args) throws InterruptedException {
        System.out.println("=== Condition 生产者-消费者示例 ===");
        // 创建容量为3的阻塞队列
        BlockingQueue<String> queue = new BlockingQueue<>(3);
        // 启动2个生产者
        for (int i = 1; i <= 2; i++) {
            new Thread(new Producer(queue, "生产者-" + i)).start();
        }
        // 启动3个消费者
        for (int i = 1; i <= 3; i++) {
            new Thread(new Consumer(queue, "消费者-" + i)).start();
        }
        // 运行10秒后结束
        Thread.sleep(10000);
        System.out.println("\n程序运行结束，队列大小: " + queue.size());
        System.exit(0);
    }
}

高级案例：多条件协作

import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.Condition;
/**
 * 多Condition协作示例 - 银行转账系统
 */
public class MultiConditionExample {
    static class BankAccount {
        private double balance;
        private final String accountName;
        private final ReentrantLock lock = new ReentrantLock();
        // 存款条件
        private final Condition sufficientFunds = lock.newCondition();
        // 取款条件
        private final Condition fundsAvailable = lock.newCondition();
        public BankAccount(String accountName, double initialBalance) {
            this.accountName = accountName;
            this.balance = initialBalance;
        }
        /**
         * 存款
         */
        public void deposit(double amount) throws InterruptedException {
            lock.lock();
            try {
                balance += amount;
                System.out.println(Thread.currentThread().getName() + 
                    " 存入 " + amount + " 到 " + accountName + 
                    ", 余额: " + balance);
                // 通知所有等待取款的线程
                fundsAvailable.signalAll();
            } finally {
                lock.unlock();
            }
        }
        /**
         * 取款
         */
        public void withdraw(double amount) throws InterruptedException {
            lock.lock();
            try {
                // 余额不足时等待
                while (balance < amount) {
                    System.out.println(Thread.currentThread().getName() + 
                        " 在 " + accountName + " 取款 " + amount + 
                        " 失败，余额不足: " + balance);
                    fundsAvailable.await();
                }
                balance -= amount;
                System.out.println(Thread.currentThread().getName() + 
                    " 从 " + accountName + " 取款 " + amount + 
                    ", 剩余余额: " + balance);
                // 通知等待存款的线程（如果需要）
                sufficientFunds.signal();
            } finally {
                lock.unlock();
            }
        }
        public double getBalance() {
            lock.lock();
            try {
                return balance;
            } finally {
                lock.unlock();
            }
        }
    }
    // 存款任务
    static class DepositTask implements Runnable {
        private final BankAccount account;
        public DepositTask(BankAccount account) {
            this.account = account;
        }
        @Override
        public void run() {
            Thread.currentThread().setName("存款线程");
            try {
                for (int i = 0; i < 5; i++) {
                    account.deposit(100);
                    Thread.sleep(800);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }
    // 取款任务
    static class WithdrawTask implements Runnable {
        private final BankAccount account;
        public WithdrawTask(BankAccount account) {
            this.account = account;
        }
        @Override
        public void run() {
            Thread.currentThread().setName("取款线程-" + 
                Thread.currentThread().getId());
            try {
                for (int i = 0; i < 3; i++) {
                    account.withdraw(200);
                    Thread.sleep(1000);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }
    public static void main(String[] args) throws InterruptedException {
        System.out.println("=== 多Condition银行转账示例 ===");
        // 创建初始余额1000的账户
        BankAccount account = new BankAccount("张三账户", 1000);
        System.out.println("初始余额: " + account.getBalance());
        // 启动存款线程
        Thread depositThread = new Thread(new DepositTask(account));
        depositThread.start();
        // 启动多个取款线程
        for (int i = 0; i < 2; i++) {
            new Thread(new WithdrawTask(account)).start();
        }
        // 等待所有线程结束
        depositThread.join();
        Thread.sleep(3000);
        System.out.println("\n最终余额: " + account.getBalance());
        System.exit(0);
    }
}

公平锁和超时等待示例

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
/**
 * Condition高级特性示例
 */
public class ConditionAdvancedExample {
    static class TimeoutQueue {
        private final int[] buffer = new int[5];
        private int count = 0;
        private int putIndex = 0;
        private int takeIndex = 0;
        private final ReentrantLock lock = new ReentrantLock(true); // 公平锁
        private final Condition notFull = lock.newCondition();
        private final Condition notEmpty = lock.newCondition();
        /**
         * 带超时的入队操作
         */
        public boolean offer(int value, long timeout, TimeUnit unit) 
                throws InterruptedException {
            long nanos = unit.toNanos(timeout);
            lock.lock();
            try {
                // 队列满时，等待指定时间
                while (count == buffer.length) {
                    if (nanos <= 0) {
                        System.out.println(Thread.currentThread().getName() + 
                            " 入队超时，放弃入队");
                        return false;
                    }
                    System.out.println(Thread.currentThread().getName() + 
                        " 队列已满，等待入队... (剩余 " + 
                        TimeUnit.NANOSECONDS.toMillis(nanos) + "ms)");
                    nanos = notFull.awaitNanos(nanos); // 等待指定时间
                }
                // 入队
                buffer[putIndex] = value;
                putIndex = (putIndex + 1) % buffer.length;
                count++;
                System.out.println(Thread.currentThread().getName() + 
                    " 入队: " + value + ", 队列大小: " + count);
                notEmpty.signal();
                return true;
            } finally {
                lock.unlock();
            }
        }
        /**
         * 带超时的出队操作
         */
        public Integer poll(long timeout, TimeUnit unit) 
                throws InterruptedException {
            long nanos = unit.toNanos(timeout);
            lock.lock();
            try {
                // 队列空时，等待指定时间
                while (count == 0) {
                    if (nanos <= 0) {
                        System.out.println(Thread.currentThread().getName() + 
                            " 出队超时，返回null");
                        return null;
                    }
                    System.out.println(Thread.currentThread().getName() + 
                        " 队列为空，等待出队... (剩余 " + 
                        TimeUnit.NANOSECONDS.toMillis(nanos) + "ms)");
                    nanos = notEmpty.awaitNanos(nanos); // 等待指定时间
                }
                // 出队
                int value = buffer[takeIndex];
                takeIndex = (takeIndex + 1) % buffer.length;
                count--;
                System.out.println(Thread.currentThread().getName() + 
                    " 出队: " + value + ", 队列大小: " + count);
                notFull.signal();
                return value;
            } finally {
                lock.unlock();
            }
        }
        public int size() {
            lock.lock();
            try {
                return count;
            } finally {
                lock.unlock();
            }
        }
    }
    public static void main(String[] args) throws InterruptedException {
        System.out.println("=== Condition高级特性示例（公平锁和超时）===");
        TimeoutQueue queue = new TimeoutQueue();
        // 生产者线程 - 快速生产
        Thread producer = new Thread(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    queue.offer(i, 2, TimeUnit.SECONDS);
                    Thread.sleep(100);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }, "生产者");
        // 消费者线程 - 慢速消费
        Thread consumer = new Thread(() -> {
            try {
                for (int i = 0; i < 8; i++) {
                    Integer value = queue.poll(2, TimeUnit.SECONDS);
                    if (value == null) {
                        System.out.println("         消费者超时等待");
                    }
                    Thread.sleep(300);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }, "消费者");
        producer.start();
        consumer.start();
        // 等待线程结束
        producer.join();
        consumer.join();
        System.out.println("\n程序结束，队列大小: " + queue.size());
    }
}

Condition的核心方法：

• await(): 等待，释放锁
• await(long time, TimeUnit unit): 等待指定时间
• awaitNanos(long nanosTimeout): 等待指定纳秒数
• signal(): 唤醒一个等待线程
• signalAll(): 唤醒所有等待线程

使用要点：

1. 必须在lock.lock()和lock.unlock()之间使用
2. 使用while而不是if来判断条件（防止虚假唤醒）
3. await()会自动释放锁，唤醒后重新竞争锁
4. 多个Condition可以实现更细粒度的控制
5. 支持公平锁和非公平锁模式