并发编程 AbstractQueuedSynchronizer类共享锁之源码解析（十八）

清疚 2022-12-28 08:26 138阅读 0赞

几天分析并发编程 AbstractQueuedSynchronizer类共享锁的源码及使用：

***一、共享式同步状态获取与释放***

共享式获取与独占式获取最主要的区别在于同一时刻能否有多个线程同时 获取到同步状态。以读写为例，如果一个程序在进行读操作，那么这一时刻写操 作均被阻塞，而读操作能够同时进行。写操作要求对资源的独占式访问，而读操 作可以是共享式访问。 在 acquireShared(int arg)方法中，同步器调用 tryAcquireShared(int arg)方法尝 试获取同步状态tryAcquireShared(int arg)方法返回值为 int 类型，当返回值大于 等于 0 时，表示能够获取到同步状态。因此，在共享式获取的自旋过程中，成功 获取到同步状态并退出自旋的条件就是 tryAcquireShared(int arg)方法返回值大于 等于 0。可以看到，在 doAcquireShared(int arg)方法的自旋过程中，如果当前节点的前驱为头节点时，尝试获取同步状态，如果返回值大于等于 0，表示该次获取同步状态成功并从自旋过程中退出。 该方法在释放同步状态之后，将会唤醒后续处于等待状态的节点。对于能够 支持多个线程同时访问的并发组件（比如 Semaphore），它和独占式主要区别在 于 tryReleaseShared(int arg)方法必须确保同步状态（或者资源数）线程安全释放， 一般是通过循环和 CAS 来保证的，因为释放同步状态的操作会同时来自多个线程。

1、源码入口：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70][]

2、源码：

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

3、点击 tryAcquireShared 进入：发现是空实现，需要子类实现

protected int tryAcquireShared(int arg) {
        throw new UnsupportedOperationException();
    }

4、点击 doAcquireShared方法，发现和独占锁类似

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);
        }
    }

**二、共享式的同步工具类**

设计一个同步工具：该工具在同一时刻，只允许至多 3 个线程同时访问，超 过 3 个线程的访问将被阻塞。

1、核心代码如下：

/**
     *类说明：共享同步工具类
     */
    public class TrinityLock  implements Lock {
    
        //为n表示允许n个线程同时获得锁
        private final Sync sync = new Sync(3);
    
        private static final class Sync extends AbstractQueuedSynchronizer {
            //private static final long serialVersionUID = -7889272986162341211L;
    
            Sync(int count) {
                if (count <= 0) {
                    throw new IllegalArgumentException("count must large than zero.");
                }
                setState(count);
            }
    
            /**
             *
             * @param reduceCount  扣减个数
             * @return  返回小于0，表示当前线程获得同步状态失败
             * 大于0，表示当前线程获得同步状态成功
             */
            public int tryAcquireShared(int reduceCount) {
                for (;;) {
                    int current = getState();
                    int newCount = current - reduceCount;
                    if (newCount < 0 || compareAndSetState(current, newCount)) {
                        return newCount;
                    }
                }
            }
    
            /**
             *
             * @param returnCount 归还个数
             * @return
             */
            public boolean tryReleaseShared(int returnCount) {
                for (;;) {
                    int current = getState();
                    int newCount = current + returnCount;
                    if (compareAndSetState(current, newCount)) {
                        return true;
                    }
                }
            }
    
            final ConditionObject newCondition() {
                return new ConditionObject();
            }
        }
    
        public void lock() {
            sync.acquireShared(1);
        }
    
        public void unlock() {
            sync.releaseShared(1);
        }
    
        public void lockInterruptibly() throws InterruptedException {
            sync.acquireSharedInterruptibly(1);
        }
    
        public boolean tryLock() {
            return sync.tryAcquireShared(1) >= 0;
        }
    
        public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
            return sync.tryAcquireSharedNanos(1, unit.toNanos(time));
        }
    
        @Override
        public Condition newCondition() {
            return sync.newCondition();
        }
    }

2、测试main方法：

public class TestMyLock {
    
        public void test() {
            final Lock lock = new TrinityLock();
            class Worker extends Thread {
    
    			public void run() {
                    lock.lock();
                    System.out.println(Thread.currentThread().getName());
                    try {
                        SleepTools.second(1);
                    } finally {
                        lock.unlock();
                    }
                }
            }
            // 启动4个子线程
            for (int i = 0; i < 8; i++) {
                Worker w = new Worker();
                //w.setDaemon(true);
                w.start();
            }
            // 主线程每隔1秒换行
            for (int i = 0; i < 10; i++) {
            	SleepTools.second(1);
                //System.out.println();
            }
        }
    
        public static void main(String[] args) {
            TestMyLock testMyLock = new TestMyLock();
            testMyLock.test();
        }
    }

3、 执行结果，每次打印3个，每隔一秒打印一次：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 1][]

4、  首先，确定访问模式。TrinityLock 能够在同一时刻支持多个线程的访问，这 显然是共享式访问，因此，需要使用同步器提供的 acquireShared(int args)方法等 和 Shared 相关的方法，这就要求 TwinsLock 必须重写 tryAcquireShared(int args) 方法和 tryReleaseShared(int args)方法，这样才能保证同步器的共享式同步状态的 获取与释放方法得以执行。

其次，定义资源数。TrinityLock 在同一时刻允许至多三个线程的同时访问， 表明同步资源数为 3，这样可以设置初始状态 status 为 3，当一个线程进行获取， status 减 1，该线程释放，则 status 加 1，状态的合法范围为 0、1 和 2,3，其中 0 表示当前已经有 3 个线程获取了同步资源，此时再有其他线程对同步状态进行获 取，该线程只能被阻塞。在同步状态变更时，需要使用 compareAndSet(int expect,int update)方法做原子性保障。

最后，组合自定义同步器。前面的章节提到，自定义同步组件通过组合自定 义同步器来完成同步功能，一般情况下自定义同步器会被定义为自定义同步组件 的内部类。到此，AQS共享锁分析完成。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70]: /images/20221120/7b3d82524b8f4428bd95323d05b89952.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 1]: /images/20221120/e14f3167f0b244e0af0a53ad552362ee.png