Java内容的复习-lock和condition的使用

以前遇到一个面试题，一些农民往桶里放苹果，一些农民往桶里面拿苹果，当桶达到1000个苹果的时候不能再放了，当桶的个数少于5个的时候不能再拿了。这个例子用lock和condition可以很好的解决。condition有await方法和signal方法，当调用await方法的时候，会释放当前的锁，然后将当前线程放到condition的等待队列中。当调用signal方法时，会调用将condition等待队列中的第一个线程放到sync队列中，那样那个线程就可以继续竞争锁。Condition和object的wait，notify方法的区别是，同一个锁可以有多个条件控制。而object的wait方法则不能。

import sun.misc.Unsafe;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Created by Administrator on 8/3/14.
 */
public class Bucket<T> {
    private List<T> cache = new ArrayList<T>();
    private ReentrantLock lock = new ReentrantLock();
    private Condition putCondition = lock.newCondition();
    private Condition getCondition = lock.newCondition();
    private int capacity = 1000;
    private int minSize = 5;

    public Bucket(int capacity, int minSize) {
        if(minSize >= capacity){
            throw new IllegalArgumentException("capacity must large than minSize!");
        }
        this.capacity = capacity;
        this.minSize = minSize;
    }

    public void put(T object){
        try{
            lock.lock();
            while (cache.size() == capacity){
                putCondition.await();
            }

            cache.add(object);
            getCondition.signal();
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            lock.unlock();
        }
    }

    public T get(){
        T value = null;
        try{
            lock.lock();
            while (cache.size() <= minSize){
                getCondition.await();
            }
            System.out.println("Cache size:" + cache.size());

            value = cache.remove(0);
            putCondition.signal();
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            lock.unlock();
        }
        return value;
    }

    static class Apple{
        public String getKey() {
            return key;
        }

        public void setKey(String key) {
            this.key = key;
        }

        private String key;

        Apple(String key) {
            this.key = key;
        }
    }

    static class Provider implements Runnable{
        private Bucket<Object> bucket;
        private String name;

        public Provider(Bucket<Object> bucket,String name) {
            this.bucket = bucket;
            this.name = name;
        }

        @Override
        public void run() {
            int index = 1;
            while (true){
                Apple apple = new Apple(name + "-" + index++);
                bucket.put(apple);
                System.out.println("Add apple " + apple.getKey() + " to bucket");
            }
        }
    }


    static class Consumer implements Runnable{
        Bucket<Object> bucket;

        public Consumer(Bucket<Object> bucket) {
            this.bucket = bucket;
        }

        @Override
        public void run() {
            while (true){
                Apple apple = (Apple) bucket.get();
                System.out.println("Get apple " + apple.getKey() + " from bucket");
            }
        }
    }

    public static void main(String[] args){
        Bucket<Object> bucket = new Bucket<Object>(100,5);
        for(int i =0;i < 2;i++){
            new Thread(new Provider(bucket,"Provider" + i)).start();
        }

        for(int i =0;i < 5;i++){
            new Thread(new Consumer(bucket)).start();
        }
    }
}

    改进版,put和take可以同时运行。模拟LinkedBlockQueue

/**
 * Created by Administrator on 8/10/14.
 */

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import sun.misc.Unsafe;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Created by Administrator on 8/3/14.
 */
public class BucketEx<T> {
    static Logger logger = LoggerFactory.getLogger(BucketEx.class);
    private ReentrantLock takeLock = new ReentrantLock();
    private ReentrantLock putLock = new ReentrantLock();
    private Condition notFull = putLock.newCondition();
    private Condition notEmpty = takeLock.newCondition();
    private Node<T> head, tail;

    private AtomicInteger count = new AtomicInteger(0);
    private int capacity = 1000;
    private int minSize = 5;

    class Node<T> {
        Node<T> next;
        T item;

        Node(T item) {
            this.item = item;
        }
    }

    public BucketEx(int capacity, int minSize) {
        if (minSize >= capacity) {
            throw new IllegalArgumentException("capacity must large than minSize!");
        }
        this.capacity = capacity;
        this.minSize = minSize;
        head = tail = new Node<T>(null);
    }

    public T dequeue() {
        Node h = head;
        Node<T> first = h.next;
        h.next = h;

        T item = first.item;
        first.item = null;
        head = first;
        return item;
    }

    public void enqueue(T item) {
        tail = tail.next = new Node<T>(item);
    }

    public void signalNotEmpty() {
        takeLock.lock();
        try{
            notEmpty.signal();
        }finally {
            takeLock.unlock();
        }
    }


    public void signalNotFull() {
        putLock.lock();
        try{
            notFull.signal();
        }finally {
            putLock.unlock();
        }
    }

    public void put(T object) throws InterruptedException {
        putLock.lockInterruptibly();
        int c = -1;
        try {
            while (count.get() >= capacity) {
                notFull.await();
            }

            enqueue(object);
            c = count.getAndIncrement();
            logger.info("put " + object.toString() + " size" + c);            
            if (c < capacity) {
                notFull.signal();
            }
        }finally {
            putLock.unlock();
        }

        if (c == minSize) {
            signalNotEmpty();
        }
    }

    public T get() throws InterruptedException {
        T item = null;
        takeLock.lockInterruptibly();
        int c = -1;

        try {
            while (count.get() <= minSize) {
                notEmpty.await();
            }

            item = dequeue();
            c = count.getAndDecrement();
            logger.info("take " + item.toString() + " size" + c);
            if (c > minSize) {
                notEmpty.signal();
            }
        } finally {
            takeLock.unlock();
        }

        if (c == capacity) {
            signalNotFull();
        }
        return item;
    }

    static class Apple {
        public String getKey() {
            return key;
        }

        public void setKey(String key) {
            this.key = key;
        }

        private String key;

        Apple(String key) {
            this.key = key;
        }

        @Override
        public String toString() {
            return "Apple{" +
                    "key='" + key + '\'' +
                    '}';
        }
    }

    static class Provider implements Runnable {
        private BucketEx<Object> bucket;
        private String name;

        public Provider(BucketEx<Object> bucket, String name) {
            this.bucket = bucket;
            this.name = name;
        }

        @Override
        public void run() {
            int index = 1;
            while (true) {
                try {
                    Thread.sleep(10);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                Apple apple = new Apple(name + "-" + index++);
                try {
                    bucket.put(apple);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }


    static class Consumer implements Runnable {
        BucketEx<Object> bucket;
        String name;

        public Consumer(BucketEx<Object> bucket,String name) {
            this.bucket = bucket;
            this.name = name;
        }

        @Override
        public void run() {
            while (true) {
                try {
                    Apple apple = (Apple) bucket.get();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        BucketEx<Object> bucket = new BucketEx<Object>(1000, 5);
        for (int i = 0; i < 1; i++) {
            new Thread(new Provider(bucket, "Provider" + i), "Provider" + i).start();
        }

        Thread.sleep(10000);
        for (int i = 0; i < 5; i++) {
            new Thread(new Consumer(bucket,"Customer" + i),"Customer" + i).start();
        }
    }
}