concurrentCondition

Posted 2020-11-04 人生如逆旅，我亦是行人。

参考文档：
Java多线程系列--“JUC锁”06之 Condition条件：http://www.cnblogs.com/skywang12345/p/3496716.html

Condition介绍

Condition的作用是对锁进行更精确的控制。Condition中的await()方法相当于Object的wait()方法，Condition中的signal()方法相当于Object的notify()方法，Condition中的signalAll()相当于Object的notifyAll()方法。不同的是，Object中的wait(),notify(),notifyAll()方法是和"同步锁"(synchronized关键字)捆绑使用的；而Condition是需要与"互斥锁"/"共享锁"捆绑使用的。基于Unsafe.park()/Unsafe.unpark()实现

Condition函数列表

// 造成当前线程在接到信号或被中断之前一直处于等待状态。
void await()
// 造成当前线程在接到信号、被中断或到达指定等待时间之前一直处于等待状态。
boolean await(long time, TimeUnit unit)
// 造成当前线程在接到信号、被中断或到达指定等待时间之前一直处于等待状态。
long awaitNanos(long nanosTimeout)
// 造成当前线程在接到信号之前一直处于等待状态。
void awaitUninterruptibly()
// 造成当前线程在接到信号、被中断或到达指定最后期限之前一直处于等待状态。
boolean awaitUntil(Date deadline)
// 唤醒一个等待线程。
void signal()
// 唤醒所有等待线程。
void signalAll()

举个栗子-单条件

public class ConditionTest {
    private static Lock lock = new ReentrantLock();
    private static Condition condition = lock.newCondition();
    public static void main(String[] args) {
        ThreadA ta = new ThreadA("t");
        lock.lock(); // 获取锁
        try {
            System.out.println(Thread.currentThread().getName() + " start ta");
            ta.start();
            System.out.println(Thread.currentThread().getName() + " block");
            condition.await(); // 等待
            System.out.println(Thread.currentThread().getName() + " continue");
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            lock.unlock(); // 释放锁
        }
    }

    static class ThreadA extends Thread {
        public ThreadA(String name) {
            super(name);
        }

        public void run() {
            lock.lock(); // 获取锁
            try {
                try {
                    Thread.sleep(1000*5);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println(Thread.currentThread().getName() + " wakup others");
                condition.signal(); // 唤醒“condition所在锁上的其它线程”
            } finally {
                lock.unlock(); // 释放锁
            }
        }
    }
}

View Code

举个栗子-多条件

/**
 * 生产者消费者模型。一个lock，两个condition，一个condition控制消费者，一个condition控制生产者
 * @author BFD_526
 *
 */
public class ConditionMoreTest {
    private static BoundedBuffer bb = new BoundedBuffer();

    public static void main(String[] args) {
        new PutThread("producer").start();
        new TakeThread("consumer").start();
    }

    static class PutThread extends Thread {
        public PutThread(String name) {
            super(name);
        }

        public void run() {
            while (true) {
                try {
                    // 向BoundedBuffer中写入数据
                    Random rd = new Random();
                    bb.put(rd.nextInt(100));
                    Thread.sleep(1000*2);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    static class TakeThread extends Thread {
        public TakeThread(String name) {
            super(name);
        }

        public void run() {
            while (true) {
                try {
                    // 从BoundedBuffer中取出数据
                    Integer num = (Integer) bb.take();
                    //Thread.sleep(1000 * 3);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

class BoundedBuffer {
    final Lock lock = new ReentrantLock();
    final Condition notFull = lock.newCondition();
    final Condition notEmpty = lock.newCondition();
    List<Integer> store = new ArrayList<Integer>();
    int limitSize=5;

    public void put(Integer x) throws InterruptedException {
        lock.lock(); // 获取锁
        System.out.println("put start size:"+store.size());
        try {
            // 如果“缓冲已满”，则等待；直到“缓冲”不是满的，才将x添加到缓冲中。
            while (store.size() == limitSize) {
                System.out.println("buffer full wait " + Thread.currentThread().getName());
                notFull.await();
            } 
            // 将x添加到缓冲中
            store.add(x);
            // 唤醒take线程，因为take线程通过notEmpty.await()等待
            notEmpty.signal();
            System.out.println(Thread.currentThread().getName() + " put  " + (Integer) x);
        } finally {
            lock.unlock(); // 释放锁
        }
    }

    public Object take() throws InterruptedException {
        lock.lock(); // 获取锁
        System.out.println("take start size:"+store.size());
        try {
            // 如果“缓冲为空”，则等待；直到“缓冲”不为空，才将x从缓冲中取出。
            while (store.size() == 0)
                notEmpty.await();
            // 将x从缓冲中取出
            Object x = store.remove(0);
            // 唤醒put线程，因为put线程通过notFull.await()等待
            notFull.signal();
            // 打印取出的数据
            System.out.println(Thread.currentThread().getName() + " take " + (Integer) x);
            return x;
        } finally {
            lock.unlock(); // 释放锁
        }
    }
}

View Code