java 并发编程之同步器

 

 在讨论Java的锁的时候，有一个概念很重要，那就是同步器，同步器可以想象成一种互斥量，当一个线程在进入临界区之前先要获得互斥量，同样，在线程获得同步状态之前也是需要获取到同步器。今天看到《java并发编程的艺术》第五章的时候觉得蛮有收获，所以这一篇算是整理下看书思路。

 

首先，我们来看一下Lock的接口

 

public interface Lock {

    void lock();
    boolean tryLock();
    void unlock();
    Condition newCondition();
}

 这个接口只是提供了几个常见的方法，而没有实现，当我们需要使用锁的时候会实现这个接口的lock和unlock方法，在这些方法的实现就是需要用到队列同步器 .AbstractQueuedSynchronizer, AbstractQueuedSynchronizer这个类提供了很多的同步方法，主要可分为三类：

1，状态追踪方法

       getState()   ：获取当前状态

       setState(int )：设置当前状态

       compareAndSetState(int expect,int update);  CAS如果当前是expect值则设置其值为update，CAS具有同步功能，因为它具有volatile的读写特性，保证编译器不会对这一段代码进行重排序

2,独占锁同步

       acquire（int arg） 获取锁

       acquireInterruptibly(int arg)  中断

       tryAcquireNanos(int arg,long nanos)  尝试获取锁，并且如果超时则放弃

       

       boolean release(int arg)  释放锁

       

3,共享锁同步 共享锁和独占所方法解释差不多，只是共享锁能够同时让多个线程执行，独占锁只能单独的线程获取同步状态后执行，其余线程会进入等待队列

       acquireShared(int arg)

       acquireSharedInterruptibly(int arg)

      tryAcqiuireSharedNanos(int arg,long nanos)

 

      boolean releaseShared(int arg);      

     

AbstractQueuedSynchronizer这个同步队列在很多的并行类或者容器中都能看到身影，因为它是负责很多的同步处理的关键，一个线程在进入临界区之前需要执行acquire方法获得同步机会，而acquire方法先要tryAcquire()如果能够获取则进入执行，如果不能获取则进入等待队列（acquire有比较复杂的方法使得进入等待队列能够实现同步，而且保持FIFO的算法）。

 

我们可以自定义自己的锁和同步器，通过继承AbstractQueuedSynchronizer这个类实现自己的同步器，而因为AbstractQueuedSynchronizer的acquire和release方法的都要执行tryAcquire和tryRelease方法，因此我们只需要重写这两个方法就能实现自定义的同步算法。

 

首先，我们来实现一个自定义的独占锁

 

package com.luchi.concurrencyImpl;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.AbstractQueuedSynchronizer;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;


public class Mutex implements Lock{

	
	//自定义内部类同步器
	private static class Sync extends AbstractQueuedSynchronizer{
		
		@Override
		protected boolean isHeldExclusively(){
			
			return getState()==1;
		}
		
		@Override
		public boolean tryAcquire(int acquires){
			
			
			if(compareAndSetState(0, 1)){
				setExclusiveOwnerThread(Thread.currentThread());
				return true;
			}
			return false;
		}
		
		@Override
		
		public boolean tryRelease(int release){
			
			if(getState()==0){
				throw new IllegalMonitorStateException();
			}
			setExclusiveOwnerThread(null);
			setState(0);
			return true;
			
			
		}
		Condition newCondition(){return new ConditionObject();}
		
	}
	
	private final Sync sync=new Sync();
	
	@Override
	public void lock() {
		// TODO Auto-generated method stub
		sync.acquire(1);
	}

	@Override
	public void unlock() {
		// TODO Auto-generated method stub
		sync.release(1);
		
	}
	@Override
	public void lockInterruptibly() throws InterruptedException {
		// TODO Auto-generated method stub
		sync.acquireInterruptibly(1);
	}

	@Override
	public boolean tryLock() {
		// TODO Auto-generated method stub
		return sync.tryAcquire(1);
	}

	@Override
	public boolean tryLock(long time, TimeUnit unit)
			throws InterruptedException {
		// TODO Auto-generated method stub
		return sync.tryAcquireNanos(1,unit.toNanos(time));
	}


	@Override
	public Condition newCondition() {
		// TODO Auto-generated method stub
		return sync.newCondition();
	}

}

 

接着我们来定义一段共享锁

 

package com.luchi.concurrencyImpl;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.AbstractQueuedSynchronizer;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

public class TwinsLock implements Lock{

	private final Sync sync=new Sync(2);
	//自定义队列同步器
	private static final class Sync extends AbstractQueuedSynchronizer{
		
		
		//初始化队列同步器的
		Sync(int count){
			if(count<=0){
				throw new  IllegalArgumentException("count must larger than zero");
				
			}
			setState(count);
		}
		
		@Override
		
		public int tryAcquireShared(int reduceCount){
			
			while(true){
				int current=getState();
				int newCount=current-reduceCount;
				if(newCount<0 || compareAndSetState(current,newCount)){
					
					return newCount;
					
				}
				
			}
		}
		
		
		@Override
		public boolean tryReleaseShared(int returnCount){
			
			while(true){
				
				int current=getState();
				int newCount=current+returnCount;
				if(compareAndSetState(current, newCount)){
					return true;
				}
			}
			
		}
		
		
		
	}
	
	
	@Override
	public void lock() {
		// TODO Auto-generated method stub
		
		sync.acquireShared(1);
	}

	@Override
	public void lockInterruptibly() throws InterruptedException {
		// TODO Auto-generated method stub
		
	}

	@Override
	public boolean tryLock() {
		// TODO Auto-generated method stub
		return false;
	}

	@Override
	public boolean tryLock(long time, TimeUnit unit)
			throws InterruptedException {
		// TODO Auto-generated method stub
		return false;
	}

	@Override
	public void unlock() {
		// TODO Auto-generated method stub
		
		
		sync.releaseShared(1);
		
	}

	@Override
	public Condition newCondition() {
		// TODO Auto-generated method stub
		return null;
	}
	
	
	

}

  可以看到，无论是共享锁合适独占锁，在自定义的时候只需要重写tryAcquire（或者tryAcquireShare）方法和tryRelease（或者tryReleaseShared）方法即可，独占锁因为“互斥量”为1所以不需要初始化，共享锁则可以自定义互斥量的多少，也就是定义多少个线程能够同时进入同步区。

 

另外我们也可以看到，锁和同步器的分离降低了两者之间的耦合度，也方便程序开发人员进行模块化定义和分工，这符合软件工程的思维