AQS同步队列结构分析
原创
同步队列结构
AQS使用的同步队列是基于一种CLH锁算法来实现。
CLH锁也是一种基于链表的可扩展、高性能、公平的自旋锁，申请线程只在本地变量上自旋，它不断轮询前驱的状态，如果发现前驱释放了锁就结束自旋.
img
同步器中包含了两个节点类型的引用，一个指向头节点(head)，一个指向尾节点(tail),没有获取到锁的线程，加入到队列的过程必须保证线程安全，因此同步器提供了一个基于CAS的设置尾节点的方法
CompareAndSetTail(Node expect,Node update),它需要传递当前线程认为的尾节点和当前节点，只有设置成功后，当前节点才能正式与之前的尾节点建立关联。
image-20210113202222401
同步器队列遵循
FIFO,首节点是获取锁成功的节点,首节点的线程在释放锁时,会唤醒后续节点,而后继节点在成功获取到锁后,会把自己设置成首节点,设置首节点是由获取锁成功的线程来完成的,由于只有一个线程能成功获取到锁,所以设置首节点不需要CAS。
image-20210113202940764
AQS实现一个线程安全的计数器
自定义互斥锁
package com.rumenz.task.aqs;
import java.util.concurrent.locks.AbstractQueuedSynchronizer;
public class MyLock {
private static final Sync STATE_HOLDER = new Sync();
/**
* 通过Sync内部类来持有同步状态， 当状态为1表示锁被持有，0表示锁处于空闲状态
*/
private static class Sync extends AbstractQueuedSynchronizer {
/**
* 是否被独占， 有两种表示方式
* 1. 可以根据状态，state=1表示锁被占用，0表示空闲
* 2. 可以根据当前独占锁的线程来判断，即getExclusiveOwnerThread()!=null 表示被独占
*/
@Override
protected boolean isHeldExclusively() {
return getExclusiveOwnerThread() != null;
}
/**
* 尝试获取锁，将状态从0修改为1，操作成功则将当前线程设置为当前独占锁的线程
*/
@Override
protected boolean tryAcquire(int arg) {
if (compareAndSetState(0, 1)) {
setExclusiveOwnerThread(Thread.currentThread());
return true;
}
return false;
}
/**
* 释放锁，将状态修改为0
*/
@Override
protected boolean tryRelease(int arg) {
if (getState() == 0) {
throw new UnsupportedOperationException();
}
setExclusiveOwnerThread(null);
setState(0);
return true;
}
}
/**
* 下面的实现Lock接口需要重写的方法，基本是就是调用内部内Sync的方法
*/
public void lock() {
STATE_HOLDER.acquire(1);
}
public void unlock() {
STATE_HOLDER.release(1);
}
}测试案例
package com.rumenz.task.aqs;
import org.omg.Messaging.SYNC_WITH_TRANSPORT;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
public class LockTest {
private final static Integer clientTotal=100000;
private final static Integer threadTotal=200;
private static Count count=new Count();
private static Count unSafe=new Count();
public static void main(String[] args) throws Exception {
ExecutorService executorService = Executors.newCachedThreadPool();
final CountDownLatch countDownLatch=new CountDownLatch(clientTotal);
final Semaphore semaphore=new Semaphore(threadTotal);
for (int i = 0; i < clientTotal; i++) {
executorService.execute(()->{
try{
semaphore.acquire();
count.getIncrement();
unSafe.getUnSafeIncrement();
semaphore.release();
}catch (Exception e){
e.printStackTrace();
}
countDownLatch.countDown();
});
}
countDownLatch.await();
System.out.println("safe:"+count.getCount());
System.out.println("unSafe:"+unSafe.getCount());
executorService.shutdown();
}
}
class Count{
private MyLock myLock;
private volatile int count;
Count() {
this.myLock=new MyLock();
}
int getCount(){
return count;
}
int getIncrement(){
myLock.lock();
count++;
myLock.unlock();
return count;
}
int getUnSafeIncrement(){
count++;
return count;
}
}输出结果
safe:100000
unSafe:99995
wx.jpg
原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。
如有侵权，请联系 cloudcommunity@tencent.com 删除。
原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。
如有侵权，请联系 cloudcommunity@tencent.com 删除。
评论
登录后参与评论
推荐阅读
目录