AQS 是 AbstractQueuedSynchronizer 的简称,是用来构建锁或者其他同步组件的基础框架,它使用int成员变量表示同步状态,通过内置的 FIFO (First in first out)队列来完成资源获取线程的排队工作。
ReetrantLock、ReetrantReadWriteLock、Semaphore内部均有Sync抽象静态内部类同步实现,Sync子类实现了公平与不公平版本。
AQS是实现锁的关键,在锁的实现中聚合同步器,利用同步器实现锁的语义。可以这样理解二者之间的关系:
package com.alwyn.nettysample.synchronizer;
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 Mutex implements Lock {
private final Sync sync = new Sync();
private static class Sync extends AbstractQueuedSynchronizer {
@Override
protected boolean isHeldExclusively() {
return getState() == 1;
}
@Override
public boolean tryAcquire(int arg) {
if (compareAndSetState(0, 1)) {
setExclusiveOwnerThread(Thread.currentThread());
return true;
}
return false;
}
@Override
protected boolean tryRelease(int releases) {
if (getState() == 0) {
throw new IllegalMonitorStateException();
}
setExclusiveOwnerThread(null);
setState(0);
return true;
}
Condition newCondition() {return new ConditionObject();}
}
@Override
public void lock() {
sync.acquire(1);
}
@Override
public void lockInterruptibly() throws InterruptedException {
sync.acquireInterruptibly(1);
}
@Override
public boolean tryLock() {
return sync.tryAcquire(1);
}
@Override
public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
return sync.tryAcquireNanos(1, unit.toNanos(time));
}
@Override
public void unlock() {
sync.release(1);
}
@Override
public Condition newCondition() {
return sync.newCondition();
}
public boolean isLocked() {
return sync.isHeldExclusively();
}
public boolean hasQueuedThreads() {
return sync.hasQueuedThreads();
}
public Collection<Thread> getQueuedThreads() {
return sync.getQueuedThreads();
}
}
复制代码
package com.alwyn.nettysample.controller;
import com.alwyn.nettysample.synchronizer.Mutex;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import java.util.Collection;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
@RestController
public class HelloController {
private int count = 0;
private AtomicInteger integer = new AtomicInteger(0);
Mutex lock = new Mutex();
@RequestMapping("/hi")
public String hi() {
try {
int i = integer.incrementAndGet();
System.out.println("request count:" + i);
boolean flag = lock.tryLock(100, TimeUnit.MILLISECONDS);
if (flag) {
count++;
System.out.println("count = " + count);
Thread.sleep(10);
lock.unlock();
} else {
boolean b = lock.hasQueuedThreads();
Collection<Thread> queuedThreads = lock.getQueuedThreads();
for (Thread thread : queuedThreads) {
System.out.println(thread.getId() + "::" + thread.getName() + "::" + thread.getState() + "::" + thread.getThreadGroup());
}
System.out.println("获取锁失败:" + b);
}
} catch (InterruptedException e) {
e.printStackTrace();
}
return "say hi world";
}
@RequestMapping("/reset")
public void setValue() {
integer.set(0);
count = 0;
}
}
复制代码