Java中的多线程与锁（四）（队列同步器）-白红宇

Java中的多线程与锁（四）（队列同步器）

阅读量：4109 次

发布时间：2019-05-25

本文共 4683 字，大约阅读时间需要 15 分钟。

0. 先来分析下以共享模式获取以及释放锁的两个方法，即 boolean releaseShared(int arg) 和 void acquireShared(int arg)

先看 acquireShared 方法，该方法尝试以共享模式获取锁，其源代码如下：

public final void acquireShared(int arg) {
           if (tryAcquireShared(arg) < 0)            doAcquireShared(arg);    }

其中的 tryAcquireShared 方法需要我们根据自己的需求在子类中覆盖/实现（这和中讲到的独占锁类似），需要留意该方法对返回值的要求及约束，tryAcquireShared 方法返回一个 int 值，该返回值的不同取值表达不同的含义，该方法的声明中对返回值的描述为：“<1> 返回值为负数，表示获取锁失败，<2> 返回值为 0，表示当前线程以共享模式成功获取锁，但是后续的其它线程的获取不会成功，<3> 返回值为正数，表示当前线程以共享模式成功获取锁，且后续的其它线程也可能会获取成功”，tryAcquireShared 方法的返回值非常重要，当它的返回值为正数时，以共享模式的获取操作会从当前节点开始在同步队列中依次向后传播（propagate），以唤醒（unpark ）其它等待共享锁的线程再次尝试获取锁（即再次执行 tryAcquireShared 方法）。

如果锁获取失败，则当前线程进入 doAcquireShared(int arg) 方法，注意该方法以 tryAcquireShared 的返回值作为参数，其代码如下：

private void doAcquireShared(int arg) {
           final Node node = addWaiter(Node.SHARED);        boolean failed = true;        try {
               boolean interrupted = false;            for (;;) {
                   final Node p = node.predecessor();                if (p == head) {
                       int r = tryAcquireShared(arg);                    if (r >= 0) {
                           setHeadAndPropagate(node, r);                        p.next = null; // help GC                        if (interrupted)                            selfInterrupt();                        failed = false;                        return;                    }                }                if (shouldParkAfterFailedAcquire(p, node) &&                    parkAndCheckInterrupt())                    interrupted = true;            }        } finally {
               if (failed)                cancelAcquire(node);        }    }

可以发现，该方法与中讲的用于独占式获取锁的方法 ‘boolean acquireQueued(final Node node, int arg)’ 代码结构非常的类似，而且其思路也是相似的，不同的地方在于 setHeadAndPropagate(Node node,int r) 方法，当前线程成功获取锁时会进入 setHeadAndPropagate 方法，该方法中将当前节点设置为头节点，并且当 tryAcquireShared 方法返回正数时，会向后传播（propagate）即唤醒头节点的后继共享类型的节点，在这个向后传播的过程中，头节点位置可能频繁发生变化，导致很可能同时有多个线程试图 unpark 同一个节点，我们依次进入以下方法：acquireShared -> doAcquireShared -> setHeadAndPropagate -> doReleaseShared，最终进入 doReleaseShared 方法，该方法中使用 CAS 操作保证只会有一个线程成功执行唤醒（unpark）后继节点的操作，其代码如下图：

当头节点的位置稳定下来，执行线程才从 for 循环返回。

再看看 releaseShared 方法，该方法代码如下：

public final boolean releaseShared(int arg) {
           if (tryReleaseShared(arg)) {
               doReleaseShared();            return true;        }        return false;    }

其中 tryReleaseShared 方法同样需要我们自定义实现，且当成功释放锁时，返回 true，进而进入 doReleaseShared 方法，尝试唤醒（unpark）后继节点。此时再看 doReleaseShared 方法，在共享模式的获取和释放锁中都会调用该方法来进行唤醒后继节点（unparkSuccessor ）的操作，其中 CAS 操作（CompareAndSwap）为锁自身的多线程安全提供了最基本的‘原子操作’保障。

1. 自定义的共享锁组件，实现 tryAcquireShared(int arg) 和 tryReleaseShared(int arg) 方法

通过继承 AbstractQueuedSynchronizer 抽象类，覆盖 tryAcquireShared(int arg) 和 tryReleaseShared(int arg) 方法实现一个简单的共享锁，代码如下：

class SharedLock{
   	private final Sync sync;	private final int count; // 保存锁的初始状态值	private final List
   
     threadsHoldSharedLock; // 记录持有锁的线程对象，控制只有持有锁的线程才能释放锁	public SharedLock(int count) {
   		sync = new Sync(count);		this.count = count;		// 确保对 list 的操作多线程安全		this.threadsHoldSharedLock = Collections.synchronizedList(new ArrayList
    
     (count));	}		public void lock() {
   		sync.acquireShared(1); // 减 1 表示获取锁	}		public void unlock() {
   		sync.releaseShared(1); // 加 1 表示释放锁	}		private class Sync extends AbstractQueuedSynchronizer {
   		private static final long serialVersionUID = 1L;		public Sync(int count) {
   			super();			setState(count);		}				@Override		protected int tryAcquireShared(int arg) {
   			int currentCount = getState();			int newCount = currentCount - arg;			if(newCount < 0) {
   				return -1;			}else {
   				if (newCount <= count && compareAndSetState(currentCount, newCount)) {
   					threadsHoldSharedLock.add(Thread.currentThread());					return newCount;				}else {
   										return -1;				}			}		}		/* 确保持有锁的线程必须成功释放锁 */		@Override		protected boolean tryReleaseShared(int arg) {
   			Thread currenThread = Thread.currentThread();			for(;;) {
   				int currentCount = getState();				int newCount = currentCount + arg;				if (threadsHoldSharedLock.contains(currenThread) 						&& newCount <= count) {
   					if (!compareAndSetState(currentCount, newCount)) {
   						// 下一次尝试						// 确保持有锁的线程必须成功释放锁						continue;					}					threadsHoldSharedLock.remove(currenThread);					return true;				}				return false;			}		}	}}

可见，通过队列同步器来自定义共享锁和独占锁，其代码结构都是类似的，主要的不同在于释放锁的操作，当释放共享锁时，在一个死循环中不断尝试，直到成功为止。因为可能会有多个线程同时执行释放共享锁的操作，所以必须考虑 CAS 操作执行失败的情况，如果失败，则再次尝试，直到成功（for 循环执行的次数，取决于同时刻执行释放锁操作的线程数量，线程越多，竞争则越激烈，失败的概率则愈高）。

共享锁也可以实现独占锁的效果，例如下面的使用方式（使得只有一个线程能够共享该锁，相当于独占锁了）：

SharedLock sharedLock = new SharedLock(1);

可以将上述这种具有独占锁效果的共享锁替换第一篇文章计数器累加程序中的独占锁，并查看程序运行结果是否正确。

转载地址：http://gclsi.baihongyu.com/

你可能感兴趣的文章