JAVA中的锁（四）-LockSuport

//阻塞当前线程，当调用unpark或者当前线程被中断，才能从park方法返回
void park()
//阻塞当前线程，blocker用来标识当前线程在等待的对象(简称阻塞对象，该对象主要用于问题排查和系统监控)
park(Object blocker)
  
//阻塞当前线程，最长不超过nanos纳秒
void parkNanos(long nanos)
void parkNanos(Object blocker, long nanos)
  
//阻塞当前线程，直到deadline时间(从1970年到deadline时间的毫秒数)
void parkUntil(long deadline)
void parkUntil(Object blocker, long deadline)
  
//唤醒处于阻塞状态的线程t
void unpark(Thread t)

public static void main(String[] args) {
	Thread thread = Thread.currentThread();
	
	LockSupport.unpark(thread);
	System.out.println("aa");
	
	LockSupport.park();
	System.out.println("bb");
	
	LockSupport.park();
	System.out.println("cc");
}
//运行结果
//这段代码打印出aa和bb，不会打印cc，因为第二次调用park的时候，线程无法获取许可出现死锁。

import java.util.concurrent.locks.LockSupport;
public class LockSupportTest {
	public static void main(String[] args) throws Exception {
		t2();
	}

	public static void t2() throws Exception {
		Thread t = new Thread(new Runnable() {
			private int count = 0;

			@Override
			public void run() {
				long start = System.currentTimeMillis();
				long end = 0;

				while ((end - start) <= 1000) {
					count++;
					end = System.currentTimeMillis();
				}
				System.out.println("about 1 second after, count=" + count);

				// 等待获取许可
				LockSupport.park();
				System.out.println("thread over. " +
                                   Thread.currentThread().isInterrupted());
			}
		});

		t.start();
		Thread.sleep(2000);
		System.out.println("main thread sleep 2 second done!");

		// 中断线程
		t.interrupt();
		System.out.println("main over");
	}
}

//运行结果：
about 1 second after, count=22987103
main thread sleep 2 second done!
main over
thread over. true

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

public class ConditionTest {
	Lock lock = new ReentrantLock();
	Condition condition = lock.newCondition(); //一般会将Condition作为成员变量
	
	public static void main(String[] args) {
		
	}
	
	public void conditionWait() throws InterruptedException {
		lock.lock();
		try {
			condition.await(); //当前线程释放锁并在此等待
		} finally {
			lock.unlock();
		}
	}
	
	public void conditionSignal() {
		lock.lock();
		try {
			condition.signal(); //通知等待的某个线程，被通知的线程将从await方法返回
		} finally {
			lock.unlock();
		}
	}
}

//当前线程进入等待状态直到被通知（signal）或中断，当前线程进入运行状态且从await方法返回的情况包括：
//1.其他线程调用该Condition的signal()或者signalAll()方法，而当前线程被选中唤醒
//2.其他线程(调用interrupt()方法)中断当前线程
//如果当前线程从await方法返回，表明该线程已经获取了Condition对象所对应的锁
void await() throws InterruptedException
  
//当前线程进入等待状态直到被通知，但是该方法对中断不敏感
void awaitUninterruptibly()
  
//当前线程进入等待状态直到被通知、中断或者超时。返回值表示剩余的时间，如果在nanosTimeout纳秒之前被唤醒
//那么返回值就是(nanosTimeout-实际耗时)，如果返回值是0或负数，那么可以认定已经超时了
long awaitNanos(long nanosTimeout) throws InterruptedException
  
//当前线程进入等待状态直到被通知、中断或到某个时间。
//如果没有到指定时间就被通知则方法返回true，否则表示到了指定时间方法返回false
boolean awaitUntil(Date deadline) throws InterruptedException
  
//唤醒一个等待在Condition上的线程，该线程从等待方法返回前必须获得与Condition相关联的锁
void signal()
  
//唤醒所有等待在Condition上的线程，该线程从等待方法返回前必须获得与Condition相关联的锁
void signalAll()

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

public class BoundedQueue<E> {
	private Object[] items;
	private int addIndex, removeIndex, count;
	private Lock lock = new ReentrantLock();
	private Condition notEmpty = lock.newCondition();
	private Condition notFull = lock.newCondition();
	
	public BoundedQueue(int size) {
		items = new Object[size];
	}
	
	public void add(E e) throws InterruptedException {
		lock.lock();
		try {
			while (count==items.length) {
				notFull.await();
			}
			items[addIndex] = e;
			if (++addIndex==items.length) {
				addIndex = 0;
			}
			++count;
			notEmpty.signal();
		} finally {
			lock.unlock();
		}
	}
	
	@SuppressWarnings("unchecked")
	public E remove() throws InterruptedException {
		lock.lock();
		try {
			while (count==0) {
				notEmpty.await();
			}
			Object x = items[removeIndex];
			if (++removeIndex==items.length) {
				removeIndex = 0;
			}
			--count;
			notFull.signal();
			return (E) x;
		} finally {
			lock.unlock();
		}
	}
}

public final void await() throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
  	//当前线程加入等待队列
    Node node = addConditionWaiter();
  	//释放同步状态，也就是释放锁
    int savedState = fullyRelease(node);
    int interruptMode = 0;
  	//被唤醒后的线程将从isOnSyncQueue退出(isOnSyncQueue(node)返回true)
  	//进而调用同步器的acquireQueued()方法加入到获取同步状态的竞争中
    while (!isOnSyncQueue(node)) {
        LockSupport.park(this);
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
            break;
    }
    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
        interruptMode = REINTERRUPT;
    if (node.nextWaiter != null) // clean up if cancelled
        unlinkCancelledWaiters();
    if (interruptMode != 0)
        reportInterruptAfterWait(interruptMode);
}

public final void signal() {
  	 //调用该方法的前置条件是当前线程必须获取了锁，可以看到signal进行了isHeldExclusively检查
  	 //也就是当前线程必须是获取了锁的线程
     if (!isHeldExclusively())
         throw new IllegalMonitorStateException();
     //获取Condition等待队列中的首节点
  	 Node first = firstWaiter;
     if (first != null)
       	 //将Condition等待队列中的首节点移动到同步队列的尾部，并使用LockSupport唤醒节点中的线程
         doSignal(first);
}

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

public class BlockingQueueTest {
    //生产者
    public static class Producer implements Runnable{
        private final BlockingQueue<Integer> blockingQueue;
        private volatile boolean flag;
        private Random random;

        public Producer(BlockingQueue<Integer> blockingQueue) {
            this.blockingQueue = blockingQueue;
            flag=false;
            random=new Random();

        }
        public void run() {
            while(!flag){
                int info=random.nextInt(100);
                try {
                    blockingQueue.put(info);
                    System.out.println(Thread.currentThread().getName()+" produce "+info);
                    Thread.sleep(50);
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }               
            }
        }
        public void shutDown(){
            flag=true;
        }
    }
    //消费者
    public static class Consumer implements Runnable{
        private final BlockingQueue<Integer> blockingQueue;
        private volatile boolean flag;
        public Consumer(BlockingQueue<Integer> blockingQueue) {
            this.blockingQueue = blockingQueue;
        }
        public void run() {
            while(!flag){
                int info;
                try {
                    info = blockingQueue.take();
                    System.out.println(Thread.currentThread().getName()
                                       +" consumer "+info);
                    Thread.sleep(50);
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }               
            }
        }
        public void shutDown(){
            flag=true;
        }
    }
    public static void main(String[] args){
        BlockingQueue<Integer> blockingQueue = new LinkedBlockingQueue<Integer>(10);
        Producer producer=new Producer(blockingQueue);
        Consumer consumer=new Consumer(blockingQueue);
        //创建5个生产者，5个消费者
        for(int i=0;i<10;i++){
            if(i<5){
                new Thread(producer,"producer"+i).start();
            }else{
                new Thread(consumer,"consumer"+(i-5)).start();
            }
        }

        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        producer.shutDown();
        consumer.shutDown();

    }
}