-
Java多线程（6）：锁与AQS（下）
之前说过，AQS（抽象队列同步器）是Java锁机制的底层实现。既然它这么优秀，是骡子是马，就拉出来溜溜吧。
首先用重入锁来实现简单的累加，就像这样：
/**
* 用重入锁实现累加
*
* @author 湘王
*/
public class MyLockTest {
private final Lock lock = new ReentrantLock();
private int value;
public int getNext() {
lock.lock();
try {
value++;
} finally {
lock.unlock();
}
return value;
}
public static void main(String[] args) {
MyLockTest myLock = new MyLockTest();
for (int i = 0; i < 5; i++) {
new Thread(new Runnable() {
@Override
public void run() {
for (int i = 0; i < 5; i++) {
System.out.println(myLock.getNext());
}
}
}).start();
}
}
}
运行结果显示数据有重复：
这么简单的计算都能出现重复，这肯定是无法接受的。
再用独占锁来试试看：
/**
* 利用AQS实现自定义独占锁
*
* @author 湘王
*/
public class MyExclusiveLock implements Lock {
@Override
public void lock() {
}
@Override
public void lockInterruptibly() throws InterruptedException {
}
@Override
public boolean tryLock() {
return false;
}
@Override
public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
return false;
}
@Override
public void unlock() {
}
@Override
public Condition newCondition() {
return null;
}
}
可以看到，实现lock接口，就需要实现若干自定义的接口。然后以内部类继承AQS的方式，实现排他锁，昨天也说过，AQS中tryAcquire()和tryRelease()是一一对应的，也就是也管获取，一个管释放，所以代码是：
/**
* 内部类继承AQS的方式，实现排他锁
*/
private static class SyncHelper extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = -7666580981453962426L;
/**
* 第一个线程进来，拿到锁就返回true；后面的线程进来，拿不到锁就返回false
*/
@Override
protected boolean tryAcquire(int arg) {
// 获取资源状态
int state = getState();
if (0 == state) {// 如果没有线程拿到资源的锁
if (compareAndSetState(0, arg)) {
// 保存当前持有同步锁的线程
setExclusiveOwnerThread(Thread.currentThread());
return true;
}
} else if (Thread.currentThread() == getExclusiveOwnerThread()) {
// 如果当前线程再次进来，state + 1，可重入
// 如果这里没有这个判断，那么程序会卡死
setState(state + arg);
return true;
}
return false;
}
/**
* 锁的获取和释放需要一一对应
*/
@Override
protected boolean tryRelease(int arg) {
// 获取资源状态
int state = getState();
// 返回最后一个通过setExclusiveOwnerThread()方法设置过的线程，或者null
if (Thread.currentThread() != getExclusiveOwnerThread()) {
throw new RuntimeException();
}
setState(state - arg);
if (0 == state) {
setExclusiveOwnerThread(null);
return true;
}
return false;
}
protected Condition newCondition() {
return new ConditionObject();
}
}
然后再用AQS实现lock接口的方法：
/**
* 利用AQS实现自定义独占锁
*
* @author 湘王
*/
public class MyExclusiveLock implements Lock {
private final SyncHelper synchepler = new SyncHelper();
@Override
public void lock() {
synchepler.acquire(1);
}
@Override
public void lockInterruptibly() throws InterruptedException {
synchepler.acquireInterruptibly(1);
}
@Override
public boolean tryLock() {
return synchepler.tryAcquire(1);
}
@Override
public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
return synchepler.tryAcquireNanos(1, unit.toNanos(time));
}
@Override
public void unlock() {
synchepler.release(1);
}
@Override
public Condition newCondition() {
return synchepler.newCondition();
}
/**
* 内部类继承AQS的方式，实现排他锁
*/
private static class SyncHelper extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = -7666580981453962426L;
/**
* 第一个线程进来，拿到锁就返回true；后面的线程进来，拿不到锁就返回false
*/
@Override
protected boolean tryAcquire(int arg) {
// 获取资源状态
int state = getState();
if (0 == state) {// 如果没有线程拿到资源的锁
if (compareAndSetState(0, arg)) {
// 保存当前持有同步锁的线程
setExclusiveOwnerThread(Thread.currentThread());
return true;
}
} else if (Thread.currentThread() == getExclusiveOwnerThread()) {
// 如果当前线程再次进来，state + 1，可重入
// 如果这里没有这个判断，那么程序会卡死
setState(state + arg);
return true;
}
return false;
}
/**
* 锁的获取和释放需要一一对应
*/
@Override
protected boolean tryRelease(int arg) {
// 获取资源状态
int state = getState();
// 返回最后一个通过setExclusiveOwnerThread()方法设置过的线程，或者null
if (Thread.currentThread() != getExclusiveOwnerThread()) {
throw new RuntimeException();
}
setState(state - arg);
if (0 == state) {
setExclusiveOwnerThread(null);
return true;
}
return false;
}
protected Condition newCondition() {
return new ConditionObject();
}
}
}
然后再运行测试：
/**
* 实现Lock接口方法并运行排他锁测试
*
* @author 湘王
*/
public class MyExclusiveLockTester {
// 用自定义AQS独占锁实现
private Lock lock = new MyExclusiveLock();
private int value;
public int accmulator() {
lock.lock();
try {
++value;
} finally {
lock.unlock();
}
return value;
}
public static void main(String[] args) throws InterruptedException {
MyExclusiveLockTester test = new MyExclusiveLockTester();
for (int i = 0; i < 5; i++) {
new Thread(new Runnable() {
@Override
public void run() {
for (int i = 0; i < 5; i++) {
System.out.println(test.accmulator());
}
}
}).start();
}
}
}
可以看到，结果无论怎么样都不会再重复了。
这个只是简单的累加，接下来用AQS来实现一个实际的生活场景。比如周末带女票或男票去步行街吃饭，这时候人特别多，需要摇号，而且一次只能进去三张号（不按人头算，按叫到的号来算），该怎么实现呢？
可以顺着这个思路：摇号机虽有很多号，但它本质上是个共享资源，很多人可以共享，但是每次共享的数量有限。这其实就是个可以指定数量的共享锁而已。
既然有了思路，那接下来就好办了。
/**
* 利用AQS实现自定义共享锁
*
* @author 湘王
*/
public class MyShareLock implements Lock {
@Override
public void lock() {
}
@Override
public void lockInterruptibly() throws InterruptedException {
}
@Override
public boolean tryLock() {
return false;
}
@Override
public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
return false;
}
@Override
public void unlock() {
}
@Override
public Condition newCondition() {
return null;
}
}
还是一样实现Lock接口，但这次是用AQS实现共享锁。
/**
* 内部类继承AQS实现共享锁
*
*/
private static class SyncHelper extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = -7357716912664213942L;
/**
* count表示允许几个线程能同时获得锁
*/
public SyncHelper(int count) {
if (count <= 0) {
throw new IllegalArgumentException("锁资源数量必须大于0");
}
// 设置资源总数
setState(count);
}
/**
* 一次允许多少个线程进来，允许数量的线程都能拿到锁，其他的线程进入队列
*/
@Override
protected int tryAcquireShared(int acquires) {
// 自旋
for (;;) {
int state = getState();
int remain = state - acquires;
// 判断剩余锁资源是否已小于0或者CAS执行是否成功
if (remain < 0 || compareAndSetState(state, remain)) {
return remain;
}
}
}
/**
* 锁资源的获取和释放要一一对应
*/
@Override
protected boolean tryReleaseShared(int releases) {
// 自旋
for (;;) {
// 获取当前state
int current = getState();
// 释放状态state增加releases
int next = current + releases;
if (next < current) {// 溢出
throw new Error("Maximum permit count exceeded");
}
// 通过CAS更新state的值
// 这里不能用setState()
if (compareAndSetState(current, next)) {
return true;
}
}
}
protected Condition newCondition() {
return new ConditionObject();
}
}
然后再来改造之前实现的接口：
/**
* 利用AQS实现自定义共享锁
*
* @author 湘王
*/
public class MyShareLock implements Lock {
public static int count;
private final SyncHelper synchepler = new SyncHelper(count);
@Override
public void lock() {
synchepler.acquireShared(1);
}
@Override
public void lockInterruptibly() throws InterruptedException {
synchepler.acquireSharedInterruptibly(1);
}
@Override
public boolean tryLock() {
return synchepler.tryAcquireShared(1) > 0;
}
@Override
public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
return synchepler.tryAcquireSharedNanos(1, unit.toNanos(time));
}
@Override
public void unlock() {
synchepler.releaseShared(1);
}
@Override
public Condition newCondition() {
return synchepler.newCondition();
}
/**
* 内部类继承AQS实现共享锁
*
*/
private static class SyncHelper extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = -7357716912664213942L;
/**
* count表示允许几个线程能同时获得锁
*/
public SyncHelper(int count) {
if (count <= 0) {
throw new IllegalArgumentException("锁资源数量必须大于0");
}
// 设置资源总数
setState(count);
}
/**
* 一次允许多少个线程进来，允许数量的线程都能拿到锁，其他的线程进入队列
*/
@Override
protected int tryAcquireShared(int acquires) {
// 自旋
for (;;) {
int state = getState();
int remain = state - acquires;
// 判断剩余锁资源是否已小于0或者CAS执行是否成功
if (remain < 0 || compareAndSetState(state, remain)) {
return remain;
}
}
}
/**
* 锁资源的获取和释放要一一对应
*/
@Override
protected boolean tryReleaseShared(int releases) {
// 自旋
for (;;) {
// 获取当前state
int current = getState();
// 释放状态state增加releases
int next = current + releases;
if (next < current) {// 溢出
throw new Error("Maximum permit count exceeded");
}
// 通过CAS更新state的值
// 这里不能用setState()
if (compareAndSetState(current, next)) {
return true;
}
}
}
protected Condition newCondition() {
return new ConditionObject();
}
}
}
接下来就该测试咱们需要的效果是否能实现了：
public class MyShareLockTester {
public static void main(String[] args) throws InterruptedException {
// 用自定义AQS共享锁实现
// 一次允许发放三把锁
MyShareLock.count = 3;
final Lock lock = new MyShareLock();
// 模拟20个客户端访问
for (int i = 0; i < 20; i++) {
new Thread(new Runnable() {
@Override
public void run() {
try {
lock.lock();
System.out.println("持有 " + Thread.currentThread().getName() + " 的客人可以进餐厅就餐");
// 每两次叫号之间间隔一段时间，模拟真实场景
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
// 使用完成释放锁
lock.unlock();
}
}
}).start();
}
}
}
这里有20个号，每次只能发放3张，运行之后就可以看到确实如此。
AQS是个很神奇也很好玩的东西，就像它的作者（也是除了高司令就是对Java影响最大的那个人，整个Java的多线程juc包代码就是他编写的）Doug Lea在AbstractQueuedSynchronizer的注释中所说：AQS只是一个框架，至于怎么玩，就是你的事了！
出处：https://www.cnblogs.com/xiangwang1111/p/16842277.html
最新更新
Java多线程（7）：JUC（上）
Java多线程（6）：锁与AQS（下）
Java多线程（6）：锁与AQS（中）
Java多线程（6）：锁与AQS（上）
Java多线程（5）：CAS
Java多线程（4）：ThreadLocal
Java多线程（3）：ThreadPool（下）
Java多线程（3）：ThreadPool（中）
Java多线程（3）：ThreadPool（上）
Java多线程（2）：线程关键字
三大常用数据库事务详解之三：事务运行
三大常用关系型数据库事务详解之二：基
三大关系型数据库事务详解之一：基本概
MongoDB常用命令（2）
MongoDB基本介绍与安装（1）
SQLServer触发器调用JavaWeb接口
SQL Server索引的原理深入解析
SqlServer2016模糊匹配的三种方式及效率问题
SQL中Truncate的用法
sqlserver 多表关联时在where语句中慎用tri
在vscode中使用R时，用快捷键来快捷键入卡
VB.NET中如何快速访问注册表
ASP.NET中图象处理过程详解
Vue（1）Vue安装与使用
JavaScript 语言入门
js将一段字符串的首字母转成大写
纯原生html编写的h5视频播放器
H5仿原生app短信验证码vue2.0组件附源码地
TypeScript（4）接口
TypeScript（3）基础类型
