Java中的锁以及并发工具类

概念上见名知义：

• 独占锁（排他锁）VS共享锁
• 公平锁VS公平锁
• 乐观锁VS悲观锁
• 可重入锁
• 自旋锁

JDK中对锁的实现

Lock接口

• ReentrantLock互斥锁（重入锁）实现类（逻辑实现在内部类Sync中）

Sync父类AbstractQueueSynchronizer队列同步器AQS

• ReadWriteLock接口，实现类为ReentrantReadWriteLock

Lock readLock()

Lock writeLock()

• Condition接口，配合Lock进行线程的阻塞和唤醒

通过Lock实现类的对象创建，newCondition()

await()

signal()

Lock lock = new ReentrantLock();
Condition condition = lock.newCondition();
public void conditionWait() throws InterruptedException{
	lock.lock();
  try{
  	condition.await();
  }finally{
  	lock.unlock();
  }
}
public void conditionSignal() throws InterruptedException{
	lock.lock();
  try{
  	condition.signal();
  }finally{
  	lock.unlock();
  }
}

Object监视器方法和Condition接口的对比：

总结：

ReentrantLock：读读互斥，读写互斥，写写互斥

ReentrantReadWriteLock：读读不互斥，读写互斥，写写互斥

引入JDK8新增的StampedLock

StampedLock：读读不互斥，读写不互斥，写写互斥

====================================================

除了上述锁与Condition搭配的同步方式

其他同步工具类如下：

Semaphore（也是基于AQS）:控制同时访问特定资源的线程数量，协调各个线程保证合理使用公共资源

主要方法：

intavailablePermits() 返回此信号量中当前可用的许可证数

intgetQueueLength() 返回正在等待获取许可的线程数

booleanhasQueuedThreads() 是否由线程正在等待获取许可

void reducePermits(int reduction) 减少reduction个许可，protected修饰

Collection getQueuedThreads() 返回所有等待获取许可的线程集合，protected修饰

Semaphore s = new Semaphore(10,true);//10个共享资源，公平
s.acquire();//每次获取一个，如果获取不到线程会阻塞
s.release();//释放

public class SemaphoreTest{
  private static final int HTREAD_COUNT = 30;
  private static ExecutorService threadPool = Executors.newFixedThreadPool(HTREAD_COUNT); 
  
  //控制10个线程的流量
  private static Semaphore s = new Semaphore(10);
  public static void main(String[] args){
  	for(int i=0;i<HTREAD_COUNT;i++){
		threadPool.execute(new Runnable(){
			public void run(){
				try{
				  s.acquire();//获取许可
				  System.out.println("save");
				  s.release();//归还许可
				}catch(){
				}
			}
		});
	}
	
	threadPool.shutdown();
  }
}

CountDownLatch（也是基于AQS）：等待多线程完成

//一个主线程要等待10个work线程全部执行结束后才能退出
CountDownLatch cdl = new CountDownLatch(10);
cdl.await();//主线程调用该方法，阻塞
cdl.countDown();//每个work执行完毕都调用1次，当减到0，主线程被唤醒

public class CountDownLatchTest{
  
  	//等待2个节点完成
	static CountDownLatch c = new CountDownLatch(2);
  public static void main(String[] args)throws InterruptedException{
  	new Thread(new Runnable(){
		@Override
	  	public void run(){
			System.out.println(1);
		  	c.countDown();//每调用一次，节点减1
		  	System.out.println(2);
		}
	}).start();
	
	c.await();//阻塞线程
	System.out.println("3");
  }
}

CylicBarrier（基于ReentrantLock+Condition实现）：可循环使用屏障

来了10个线程，这10个线程互相等待，到齐后一起被唤醒，各自执行接下来的逻辑；然后，这10个线程继续互相等待，到齐后再一起 被唤醒。每一轮被称为一个Generation，就是一次同步点（屏障点）。

public class CyclicBarrierTest{
	static CyclicBarrier c = new CyclicBarrier(2);//屏障拦截的线程数量是2
  public static void main(String[] args){
  	new Thread(
		@Override
	    public void run(){
			try{
				c.await();//已经到达了屏障，当前线程被阻塞
			}catch(Exception e){
			}
		  
		  	System.out.println(1);
		} 
	).start();
	
	//如果构造改成3，则主子线程都会等待，因为没有第三个线程执行await到达屏障
	//因为主线程和子线程的调度是由CPU决定的，两个线程都有可能先执行，所以会产生两种输出(1 2 或 2 1)
	try{
		c.await();
	}catch(Exception e){
	}
	System.out.println(2);
  }
}

public class CyClicBarrierTest{
  //该构造在线程到达屏障时，优先执行barrierAction
  //拦截线程数量是2，必须等待第一个线程和A都执行完，再执行主线程（先执行主线程）
	static CyclicBarrier c = new CyclicBarrier(2,new A());
	public static void main(String[] args){
		new Thread(new Runnable(){
			@Override
		  public void run(){
		  	try{
				c.await();
			}catch(Exception e){
			}
			System.out.println(1);
		  }
		}).start();
	  
	  try{
	  	c.await();
	  }catch(Exception e){
	  }
	  System.out.println(2);
	}
  
  static class A implements Runnable{
  	@Override
	public void run(){
		System.out.println(3);  //结果 3 1 2
	}
  }
}

public clas BankWaterService implements Runnable{
	//创建四个屏障,优先执行当前类的线程任务
  private CyclicBarrier c = new CyclicBarrier(4.this);
	
  //创建四个线程的线程池
  private Executor executor = Executors.newFixedThreadPool(4);
  private ConcurrentHashMap<String,Integer> map = new ConcurrentHashMap<>();
  
  private void count(){
  	for(int i=0;i<4;i++){
		executor.execute(new Runnable(){
			public void run(){
				map.put(Thread.currentThread().getName(),1);
			  c.await();//省略异常处理，阻塞
			}
		});
	}
  }
  
  //将四个线程的结果进行汇总
 	public void run(){
		int result = 0;
	  for(Entry<String,Integer> s:map.entrySet()){
	  	result += s.getValue();
	  }
	  map.put("result",result);
	  System.out.println(result);
	}
  
  public static void main(String[] args){
  	BankWaterService bs = new BankWaterService();
	bs.count();
  }
}

public class CyclicBarrierTest{
	static CyClicBarrier c = new CyclicBarrier(2);
  public static void main(String[] args){
  	Thread t = new Thread(new Runnable(){
		public void run(){
			c.await();//阻塞（省略处理异常）
		}
	});
	t.start();
	t.interrupt();
	
	try{
	c.await();
	}catch(Exception e){
	  //isBroken()方法用来了解阻塞的线程是否被中断
		System.out.println(c.isBoken());//true
	}
  }
}

Exchanger

当一个线程调用exchange准备和其他线程交换数据的时候，一种是没有其他线程要交换数据，自己只能自旋或者阻塞，等待；另一种是恰好有其他线程在 Slot 里面等着，那么和对方交换。

Exchange<String> ex = new Exchange<>();

ThreadA a = new ThreadA(ex);

public class ExchagerTest{
	private static final Exchange<String> e = new Exchanger<>();
 	private static ExecutorService threadPool = Executors.newFixedThreadPool(2);
  	public static void main(String[] args){
		threadPool.execute(new Runnable(){
			public void run(){
				String a = "银行流水A"
				e.exchange(a);//交换数据（省略处理异常）
			}
		});
	  
	  threadPool.execute(new Runnable(){
	  	public void run(){
			String b = "银行流水B"
			e.exchange(b);//如果两个线程有一个没有执行exchange()方法，则会一直等待
		}
	  });
	}
}

Phaser（替代CyclicBarrier和CountDownLatch）

Phaser ph = new Phaser(10);
ph.awaitAdance(ph.getPhase());//阻塞，等待当前同步节点完成
ph.arrive();//每个线程完成任务调用1次，表达同步到达了该同步点

Phaser ph = new Phaser(10);
ph.arriveAndAwaitAdvance();//第一个同步点

ph.arriveAndAwaitAdvance();//第二个同步点