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; } } 复制代码