Java线程同步的Monitor机制(Lock配合Condition)

Monitor模式是一种常见的并行开发机制, 一个Monitor实例能够被多个线程安全使用, 全部的monitor下面的方法在运行时是互斥的, 这种互斥机制机制能够用于一些特性, 例如让线程等待某种条件, 在等待时线程会将CPU时间交出去, 可是在条件知足时确保从新得到CPU时间. 在条件达成时, 你能够同时通知一个或多个线程. 这样作有如下的优势:

1. 全部的同步代码都集中在一块儿, 用户不须要知道这是如何实现的
2. 代码不依赖于线程数量, 线程数量只取决于业务须要
3. 不须要对某个互斥对象作释放, 不存在忘记的风险

一个Monitor的结构是这样的

public class SimpleMonitor {
    public method void testA(){
        //Some code
    }

    public method int testB(){
        return 1;
    }
}

使用Java代码不能直接建立一个Monitor, 要实现Monitor, 须要使用Lock和Condition类. 通常使用的Lock是ReentrantLock, 例如

public class SimpleMonitor {
    private final Lock lock = new ReentrantLock();

    public void testA() {
        lock.lock();

        try {
            //Some code
        } finally {
            lock.unlock();
        }
    }

    public int testB() {
        lock.lock();

        try {
            return 1;
        } finally {
            lock.unlock();
        }
    }
}

若是不须要判断条件, 那么用synchronized就能够了. 在须要判断条件的状况下, 使用Lock的newCondition()方法建立Condition, 能够经过Condition的await方法, 让当前线程wait, 放弃cpu时间. 而后用signal或者signalAll方法让线程从新得到CPU时间. signalAll方法会唤起全部wait在当前condition的线程. 下面是一个例子, 一个须要被多个线程使用的容量固定的buffer.

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

public class BoundedBuffer {
    private final String[] buffer;
    private final int capacity;

    private int front;
    private int rear;
    private int count;

    private final Lock lock = new ReentrantLock();

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

    public BoundedBuffer(int capacity) {
        super();

        this.capacity = capacity;

        buffer = new String[capacity];
    }

    public void deposit(String data) throws InterruptedException {
        lock.lock();

        try {
            while (count == capacity) {
                notFull.await();
            }

            buffer[rear] = data;
            rear = (rear + 1) % capacity;
            count++;

            notEmpty.signal();
        } finally {
            lock.unlock();
        }
    }

    public String fetch() throws InterruptedException {
        lock.lock();

        try {
            while (count == 0) {
                notEmpty.await();
            }

            String result = buffer[front];
            front = (front + 1) % capacity;
            count--;

            notFull.signal();

            return result;
        } finally {
            lock.unlock();
        }
    }
}

代码说明

1. 这两个方法经过lock互斥
2. 而后经过两个condition变量, 一个用于在buffer非空时等待, 一个用于buffer未满时等待
3. 上面使用while循环将await包围, 这是为了防止在使用Signal&Condition时产生signal stealers问题.
4. 以上方法能够安全地在多个线程中被调用

还有一个例子, 用于协调多个线程按固定顺序进行输出

public class DemoLockCondition {
    private final Lock lock = new ReentrantLock();
    private final Condition one = lock.newCondition();
    private final Condition two = lock.newCondition();
    private final Condition three = lock.newCondition();

    public void action1() {
        while (true) {
            lock.lock();
            System.out.println("1");
            try {
                two.signal();
                one.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                lock.unlock();
            }
        }
    }

    public void action2() {
        while (true) {
            lock.lock();
            System.out.println("2");
            try {
                three.signal();
                two.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                lock.unlock();
            }
        }
    }

    public void action3() {
        while (true) {
            lock.lock();
            System.out.println("3");
            try {
                one.signal();
                three.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                lock.unlock();
            }
        }
    }

    public static void main(String[] args) {
        DemoLockCondition demo = new DemoLockCondition();
        new Thread(()->demo.action1()).start();
        new Thread(()-> demo.action2()).start();
        new Thread(()-> demo.action3()).start();
    }
}

若是是使用wait()和notify()的话, 就要写成这样, 这种状况下, 运行时notify()随机通知的线程, 是有可能不知足而跳过的.

public class DemoThreadWait2 {
    private Object obj = 0;
    private int pos = 1;

    public void one(int i) {
        synchronized (obj) {
            if (pos == i) {
                System.out.println("T-" + i);
                pos = i % 3 + 1;
            } else {
                // System.out.println(".");
            }
            obj.notify();
            try {
                obj.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) {
        DemoThreadWait2 demo = new DemoThreadWait2();
        new Thread(()->{
            while(true) {
                demo.one(1);
            }
        }).start();

        new Thread(()->{
            while(true) {
                demo.one(2);
            }
        }).start();

        new Thread(()->{
            while(true) {
                demo.one(3);
            }
        }).start();
    }
}