同步工具类（锁闭锁栅栏信号量）

Posted 2020-12-21 mougg

目录

• 1 AQS
• 2 ReentrantLock
  • 2.1 使用场景
  • 2.2 AQS 实现
  • 2.3 分布式实现
• 3 CountDownLatch
  • 3.1 使用场景
  • 3.2 AQS 实现
  • 3.3 Redis 实现
• 4 CylicBarrier
  • 4.1 使用场景
  • 4.2 AQS 实现
  • 4.3 Redis 实现
• 5 Semaphore
  • 5.1 使用场景
  • 5.2 AQS 实现
  • 5.3 Redis 实现

1 AQS

state AQS 的关键
CAS Compare And Swap 修改 state
Unsafe 使用 LockSupport 的 park()、unpark() 挂起和唤醒线程
同步队列 sync queue
条件队列 condition queue
等待队列 还有什么阻塞队列，看看哪个是真的

2 ReentrantLock

2.1 使用场景

2.2 AQS 实现

ReadWriteReentrantLock
StampLock

2.3 分布式实现

参考 分布式锁

3 CountDownLatch

闭锁，允许N个线程一直等待，直到其他线程执行的操作全部完成。

CountDownLatch(int count)

count 表示计数器。

3.1 使用场景

数据库脱敏，每批次查询 1000 条，提交给线程池，线程在脱敏完成后 countDown，然后继续。
主线程 await 阻塞，等待脱敏完 1000 条记录后返回，并批量更新数据库。

3.2 AQS 实现

1. 主线程构造 CountDownLatch，设置 AQS state 为 count
2. 主线程调用 await()，当 state > 0 时进入 AQS 阻塞队列
3. 子线程调用 countDown() 原子递减 state，当 state == 0 时唤醒所有调用await()方法阻塞的线程

3.3 Redis 实现

setCount() EXISTS key，如果不存在则 SET count
countDown() DECR key，如果返回值 <= 0，则 DEL key
await() 在 while 循环中 GET key，如果返回值 == 0，则结束循环

4 CylicBarrier

栅栏，让一组线程互相等待，直到所有线程都到达一个同步点。Cylic 表示可以循环利用。

CyclicBarrier(int parties, Runnable barrierAction)

parties 表示预期到达栅栏的线程数；barrierAction 表示所有线程到达栅栏后执行的回调

4.1 使用场景

数据库脱敏，每批次查询 1000 条，提交给线程池，线程在脱敏完成后 await。
当 1000 个线程 await 时，CylicBarrier 唤醒其中一个线程执行批量更新。

4.2 AQS 实现

1. 主线程构造 CylicBarrier，设置 AQS state 为 parties
2. 子线程调用 await()，获取排斥锁，递减 state，进入阻塞队列，释放锁
3. 当 state == 0 时，唤醒其中一个线程执行构造器中设置的回调，并重置 state 为 parties，循环利用

4.3 Redis 实现

5 Semaphore

信号量，控制资源可被同时访问的线程个数。

Semaphore(int permits, boolean fair)

permits 表示许可个数；fair 表示公平竞争，默认 false。

5.1 使用场景

限流。

5.2 AQS 实现

1. 主线程构造 Semaphore，设置 AQS state 为 permits
2. 主线程调用 acquire(N)，用 state - N，如果小于0，则进入阻塞队列，大于0则通过CAS设置当前信号量为剩余值，同时返回剩余值
3. 子线程调用 release(N)，用 state + N，同时不停的尝试因为调用acquire()进入阻塞的线程

5.3 Redis 实现

setPermits() GET key，如果返回值 == 0，则 SET key
acquire() GET key，如果返回值 > 0，则 INCRBY permits
release() INCRBY permits