手把手教你构建源码级组件——Java指定共享线程数目的共享锁

2023-05-16

文章目录

构造同步组件的步骤
- 1. 定义内部类Syn
- 2. 继承同步器，重写指定方法
- 3. 调用同步器方法
指定共享线程数目的共享锁实现
- 代码实现
- 测试Demo
- 运行结果
- 结果分析

构造同步组件的步骤

之前的学习中我们学习了AQS的原理，其中有许多构建锁与同步器的相关概念我们需要了解到：

首先同步器是实现锁的关键，在锁的实现中聚合同步器，利用同步器实现锁的语义；
锁是面向使用者的，提供锁交互的实现；
同步器是面向锁的实现者，简化了锁的实现方式，屏蔽了同步状态管理、线程排队、等待/唤醒等底层操作。

从代码层面，同步器是基于模板模式实现的，可以通过AQS可重写的方法进行子类具体功能实现：

例如下面是AQS中tryAcquire模板方法的源码（如果子类没实现会怕抛出异常）

/**
* 模板方法：
*  protected关键字
*  没有任何实现
* @param arg
* @return
*/
protected boolean tryAcquire(int arg) {
    throw new UnsupportedOperationException();
}

那么我们在构建同步组件的时候也就是需要实现以下几步：

1. 定义内部类Syn

随后将同步器组合在自定义同步组件的实现中，即定义内部类Syn继承AQS

public class XXX implements Lock {
    public class Sync extends AbstractQueuedSynchronizer{
	}
}

2. 继承同步器，重写指定方法

之后在Syn中重写AQS方法，根据同步器需求如下挑选实现不同方法

tryAcquire(int arg)：独占式获取同步状态；
tryRelease(int arg)：独占式释放同步状态；
tryAcquireShared(int arg)：共享式获取同步状态，返回大于0的值表示获取成功，否则失败
tryReleaseShared(int arg)：共享式释放锁
isHeldExclusively()：当前线程是否在独占模式下被线程占用，一般该方法表示是否被当前线程占用

例如不可重入同步器：

public class XXX implements Lock {
	public class Sync extends AbstractQueuedSynchronizer{
	
        @Override
        protected boolean tryAcquire(int arg) {
            final Thread current = Thread.currentThread();
            if (compareAndSetState(0, 1)) {
                // 获取成功之后，当前线程是该锁的持有者，不需要再可重入数
                setExclusiveOwnerThread(current);
                return true;
            }
            return false;
        }

        @Override
        protected boolean tryRelease(int arg) {
            if (getState() == 0) {
                throw new IllegalMonitorStateException();
            }
            setExclusiveOwnerThread(null);
            setState(0);
            return true;
        }

        @Override
        protected boolean isHeldExclusively() {
              return getState() == 1;
        }
        // 返回Condition，每个Condition都包含了一个队列
        Condition newCondition() {
            return new ConditionObject();
        }
    }
}

3. 调用同步器方法

最后调用同步器提供的模板方法，即同步组件类实现Lock方法之后，在lock/unlock方法中调用内部类Syn的方法acquire(int arg)等方法

public class XXX implements Lock {
    
   ........
   	private final Sync sync = new Sync();
    @Override
    public void lock() {
        sync.acquire(1);
    }
    @Override
    public void unlock() {
        sync.release(1);
    }
    ........

}

具体请看下面的实验部分

指定共享线程数目的共享锁实现

代码实现

package com.yyl.threadtest.utils;

import java.util.Collection;
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 MyShareLock implements Lock {

    // 可以看到共享等待队列中的线程
    public Collection<Thread> getSharedQueuedThreads(){
        return syn.getSharedQueuedThreads();
    }
    private final Syn syn = new Syn(3);

    private static final class Syn extends AbstractQueuedSynchronizer {
        int newShareCount=0;
        Syn(int shareCount){
            if (shareCount <= 0) {
                throw new IllegalArgumentException("share count must large than zero");
            }
            // 设置初始共享同步状态
            setState(shareCount);
        }

        /**
         * 共享锁指定数目
         * @param reduceShareCount
         * @return
         */
        @Override
        protected int tryAcquireShared(int reduceShareCount) {

            for (;;){
                int currentShareCount = getState();
                newShareCount = currentShareCount- reduceShareCount;
                if (newShareCount < 0 ||
                        compareAndSetState(currentShareCount,newShareCount)) {
                    // newShareCount大于等于０才说明获取锁成功
                    if (newShareCount >= 0) {
//                        System.out.println(Thread.currentThread().getName()+" hold lock, current share count is "+newShareCount+", "+new Date());
                    }
                    // newShareCount小于0表示获取失败所以需要返回
                    // compareAndSetState(currentShareCount,newShareCount)为true自然表示成功需要返回
                    return newShareCount;
                }
            }
        }

        @Override
        protected boolean tryReleaseShared(int returnShareCount) {
            for (;;){
                int currentShareCount = getState();
                newShareCount = currentShareCount + returnShareCount;
                if (compareAndSetState(currentShareCount,newShareCount)) {
                // System.out.println(Thread.currentThread().getName() +" release lock, current share count is "+newShareCount+", "+new Date());
                    return true;
                }
            }
        }
        protected int getShareCount(){
            return getState();
        }
    }

    /**
     * 调用内部同步器Syn的acquireShare方法
     */
    @Override
    public void lock() {
        syn.acquireShared(1);
    }
    /**
     * 调用内部同步器Syn的releaseShared方法
     */
    @Override
    public void unlock() {
        syn.releaseShared(1);
    }

    @Override
    public void lockInterruptibly() throws InterruptedException {
        if (Thread.interrupted()) {
            throw new IllegalStateException();
        }
        syn.acquireInterruptibly(1);
    }

    @Override
    public boolean tryLock() {
        return false;
    }

    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        return false;
    }

    @Override
    public Condition newCondition() {
        return null;
    }
}

测试Demo

package com.yyl.threadtest.utils;

import java.util.Date;

public class ShareLockTest {
    public static void main(String[] args) {
        final MyShareLock lock = new MyShareLock();
        class Worker extends Thread {
            @Override
            public void run() {
                // 一直不停在获取锁
                while (true) {
                    lock.lock();
                    try {
                        System.out.println(Thread.currentThread().getName() +" hold lock, "+new Date());
                        // System.out.println(lock.getSharedQueuedThreads());
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } finally {
                        lock.unlock();
                        System.out.println(Thread.currentThread().getName() +" release lock, "+new Date());
                    }
                }
            }

        }
        for (int i = 0; i < 10; i++) {
            Worker worker = new Worker();
            // 以守护进程运行，VM退出不影响运行，这里只是为了一个打印效果，去掉注释一直打印
            worker.setDaemon(true);
            worker.start();
        }
        // 每隔一秒换行
        for (int j = 0; j < 10; j++) {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println();
        }
    }
}

运行结果

Thread-0 hold lock, Thu Feb 23 12:55:28 CST 2023
Thread-1 hold lock, Thu Feb 23 12:55:28 CST 2023
Thread-2 hold lock, Thu Feb 23 12:55:28 CST 2023

Thread-0 release lock, Thu Feb 23 12:55:29 CST 2023
Thread-4 hold lock, Thu Feb 23 12:55:29 CST 2023
Thread-5 hold lock, Thu Feb 23 12:55:29 CST 2023
Thread-1 release lock, Thu Feb 23 12:55:29 CST 2023
Thread-3 hold lock, Thu Feb 23 12:55:29 CST 2023
Thread-2 release lock, Thu Feb 23 12:55:29 CST 2023

Thread-5 release lock, Thu Feb 23 12:55:31 CST 2023
Thread-7 hold lock, Thu Feb 23 12:55:31 CST 2023
Thread-3 release lock, Thu Feb 23 12:55:31 CST 2023
Thread-9 hold lock, Thu Feb 23 12:55:31 CST 2023
Thread-4 release lock, Thu Feb 23 12:55:31 CST 2023
Thread-6 hold lock, Thu Feb 23 12:55:31 CST 2023

Thread-8 hold lock, Thu Feb 23 12:55:32 CST 2023
Thread-6 release lock, Thu Feb 23 12:55:32 CST 2023
Thread-9 release lock, Thu Feb 23 12:55:32 CST 2023
Thread-7 release lock, Thu Feb 23 12:55:32 CST 2023
Thread-1 hold lock, Thu Feb 23 12:55:32 CST 2023
Thread-0 hold lock, Thu Feb 23 12:55:32 CST 2023

Thread-8 release lock, Thu Feb 23 12:55:33 CST 2023
Thread-3 hold lock, Thu Feb 23 12:55:33 CST 2023
Thread-2 hold lock, Thu Feb 23 12:55:33 CST 2023
Thread-1 release lock, Thu Feb 23 12:55:33 CST 2023
Thread-5 hold lock, Thu Feb 23 12:55:33 CST 2023
Thread-0 release lock, Thu Feb 23 12:55:33 CST 2023

Thread-5 release lock, Thu Feb 23 12:55:34 CST 2023
Thread-6 hold lock, Thu Feb 23 12:55:34 CST 2023
Thread-9 hold lock, Thu Feb 23 12:55:34 CST 2023
Thread-4 hold lock, Thu Feb 23 12:55:34 CST 2023
Thread-3 release lock, Thu Feb 23 12:55:34 CST 2023
Thread-2 release lock, Thu Feb 23 12:55:34 CST 2023

Thread-4 release lock, Thu Feb 23 12:55:35 CST 2023
Thread-7 hold lock, Thu Feb 23 12:55:35 CST 2023
Thread-8 hold lock, Thu Feb 23 12:55:35 CST 2023
Thread-9 release lock, Thu Feb 23 12:55:35 CST 2023
Thread-1 hold lock, Thu Feb 23 12:55:35 CST 2023
Thread-6 release lock, Thu Feb 23 12:55:35 CST 2023

Thread-1 release lock, Thu Feb 23 12:55:36 CST 2023
Thread-8 release lock, Thu Feb 23 12:55:36 CST 2023
Thread-3 hold lock, Thu Feb 23 12:55:36 CST 2023
Thread-7 release lock, Thu Feb 23 12:55:36 CST 2023
Thread-5 hold lock, Thu Feb 23 12:55:36 CST 2023
Thread-0 hold lock, Thu Feb 23 12:55:36 CST 2023

Thread-4 hold lock, Thu Feb 23 12:55:37 CST 2023
Thread-5 release lock, Thu Feb 23 12:55:37 CST 2023
Thread-3 release lock, Thu Feb 23 12:55:37 CST 2023
Thread-0 release lock, Thu Feb 23 12:55:37 CST 2023
Thread-2 hold lock, Thu Feb 23 12:55:37 CST 2023
Thread-9 hold lock, Thu Feb 23 12:55:37 CST 2023

Thread-6 hold lock, Thu Feb 23 12:55:38 CST 2023
Thread-2 release lock, Thu Feb 23 12:55:38 CST 2023
Thread-4 release lock, Thu Feb 23 12:55:38 CST 2023
Thread-1 hold lock, Thu Feb 23 12:55:38 CST 2023
Thread-8 hold lock, Thu Feb 23 12:55:38 CST 2023
Thread-9 release lock, Thu Feb 23 12:55:38 CST 2023


Process finished with exit code 0

结果分析

该指定共享线程数量N的共享锁的最终目的就是多个线程可以持有锁（同步状态），达到共享线程数量N（代码中默认为2）时，其它线程将进入Queue等待获取同步结果，同一时刻只能最多有N个线程持有锁。

同样地，我们分析开头运行结果：

Thread-0 hold lock, Thu Feb 23 12:55:28 CST 2023
Thread-1 hold lock, Thu Feb 23 12:55:28 CST 2023
Thread-2 hold lock, Thu Feb 23 12:55:28 CST 2023

Thread-0 release lock, Thu Feb 23 12:55:29 CST 2023
Thread-4 hold lock, Thu Feb 23 12:55:29 CST 2023
Thread-5 hold lock, Thu Feb 23 12:55:29 CST 2023
Thread-1 release lock, Thu Feb 23 12:55:29 CST 2023
Thread-3 hold lock, Thu Feb 23 12:55:29 CST 2023
Thread-2 release lock, Thu Feb 23 12:55:29 CST 2023

10个线程不停竞争锁，一开始Thread-0、Thread-1、Thread-2在12:55:28时刻同时获取到了锁，此时已经达到共享数量的最大值，即N，之后持有锁1秒，Thread-0、Thread-1、Thread-2在12:55:29时刻立马释放锁，同时Thread-3、Thread-4、Thread-5立马退出等待队列立马竞争持有锁。

从结果来看，完全是符合ShareLock共享锁功能的：同一时刻最多允许N个线程持有锁，其它线程等待持有线程释放锁！

本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)

Java