AQS
可以先查看 Java并发 - 读写锁与AQS简单了解 进行简单的了解
AQS中的队列
通过维护 state 进行加锁和解锁（含读锁（共享锁）和写锁（独占锁））
/**
* The synchronization state.
*/
private volatile int state;
继承 AbstractQueuedSynchronizer 后需要实现的方法
- isHeldExclusively():该线程是否正在独占资源。只有用到condition才需要去实现它。
- tryAcquire(int):独占方式。尝试获取资源，成功则返回true，失败则返回false。
- tryRelease(int):独占方式。尝试释放资源，成功则返回true，失败则返回false。
- tyAcquireShared(int):共享方式。尝试获取资源。负数表示失败;0表示成功，但没有剩余可用资源;正数表示成功，且有剩余资源。
- tryReleaseShared(int):共享方式。尝试释放资源，如果释放后允许唤醒后续等待结点返回true，否则返回false。
1. Semaphore （依赖了 AQS 定义的模板方法）
Semaphore是一个计数信号量，常用于限制可以访问某些资源（物理或逻辑的）线程数目。
简单说，是一种用来控制并发量的共享锁。
例子
package concurrent;
import java.util.concurrent.Semaphore;
public class Semaphore_Demo {
static Semaphore sp =new Semaphore(6);
public static void main(String[] args) {
for (int i = 0; i <1000 ; i++) {
new Thread(()->{
try {
sp.acquire(); //就是拿信号量
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("当前信号量限制....");
try {
Thread.sleep(2000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("执行完了");
sp.release(); //执行完了，就释放信号量
}).start();
}
}
}
当前信号量限制....
当前信号量限制....
当前信号量限制....
当前信号量限制....
当前信号量限制....
当前信号量限制....
执行完了
执行完了
执行完了
执行完了
当前信号量限制....
执行完了
执行完了
当前信号量限制....
.....
以上例子表明，Semaphore(6) 限制了只能同时有六个线程执行，也就起到了限流的作用
- 自己实现Semaphore的例子
package concurrent;
import java.util.concurrent.locks.AbstractQueuedSynchronizer;
public class MySemaphore {
private Sync sync;
public MySemaphore(int permits) {
sync = new Sync(permits);
}
public void acquire() {
sync.acquireShared(1);
}
public void release() {
sync.releaseShared(1);
}
class Sync extends AbstractQueuedSynchronizer {
// 信号量大小
private int permits;
public Sync(int permits) {
this.permits = permits;
}
@Override
protected int tryAcquireShared(int arg) {
//定义自己的方法
int state = getState();
int nextState = state + arg;
if (nextState <= permits) {
if (compareAndSetState(state, nextState))
return 1;
}
return -1;
}
@Override
protected boolean tryReleaseShared(int arg) {
int state = getState();
if (compareAndSetState(state, state - arg)) {
return true;
} else {
return false;
}
}
}
}
执行代码
package concurrent;
public class Semaphore_Demo {
static MySemaphore sp =new MySemaphore(6);
public static void main(String[] args) {
for (int i = 0; i <1000 ; i++) {
new Thread(()->{
sp.acquire(); //就是拿信号量
System.out.println("当前信号量限制....");
try {
Thread.sleep(2000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("执行完了");
sp.release(); //执行完了，就释放信号量
}).start();
}
}
}
执行结果
当前信号量限制....
当前信号量限制....
当前信号量限制....
当前信号量限制....
当前信号量限制....
当前信号量限制....
执行完了
执行完了
执行完了
执行完了
执行完了
执行完了
2. CountDownLatch (依赖了 AQS 定义的模板方法)
即实现了一种计数器，可以认为是倒数计时，比方说 从 6 开始依次递减（此时只是准备就绪），然后到 0 的时候再一起启动。
好比有这样的一个跑步比赛，比赛开始之前，所有的选手都需要向裁判示意已经准备就绪，一共有六个选手，那么每一个选手向裁判示意准备就绪，未准备就绪的人数就减1，直到所有的选手都准备就绪，此时未准备就绪的人数为0，那么裁判就会扣下发令枪示意比赛开始。
例子
package concurrent;
import java.util.concurrent.CountDownLatch;
public class CountDownLatch_Demo {
public static void main(String[] args) throws InterruptedException {
CountDownLatch latch = new CountDownLatch(6); //计数为6
for (int i = 0; i <6 ; i++) {
new Thread(()->{
System.out.println("开始准备.....");
latch.countDown();//计数减一
}).start();
Thread.sleep(1000);
}
latch.await(); //每个线程执行一次，则-1，在latch为0的时候开始向下运行 这是这些线程都准备就绪，然后去一起干同一件事
//还有一种方式， 将一个活分为多段，每个线程去干一段
// for (int i = 0; i <6 ; i++) {
// new Thread(()->{
// latch.countDown(); // 计数减一
// try {
// latch.await(); // 阻塞 -- > 0
// System.out.println("线程："+Thread.currentThread().getName()+"执行完毕");
// } catch (InterruptedException e) {
// e.printStackTrace();
// }
// }).start();
// }
System.out.println("开始干活....");
}
}
开始准备.....
开始准备.....
开始准备.....
开始准备.....
开始准备.....
开始准备.....
开始干活....
- 自己实现CountDownLatch的例子
package concurrent;
import java.util.concurrent.locks.AbstractQueuedSynchronizer;
public class MyCountDownLatch {
private Sync sync;
public MyCountDownLatch(int count) {
sync = new Sync(count);
}
public void countDown() {
sync.releaseShared(1);
}
public void await() {
sync.acquireShared(1);
}
class Sync extends AbstractQueuedSynchronizer {
public Sync(int count) {
setState(count);
}
@Override
protected int tryAcquireShared(int arg) {
return getState() == 0 ? 1 : -1;
}
@Override
protected boolean tryReleaseShared(int arg) {
while (true) {
int c = getState();
if (c == 0)
return false;
int nextc = c - 1;
if (compareAndSetState(c, nextc)) {
return nextc == 0;
}
}
}
}
}
执行代码
package concurrent;
import java.util.concurrent.CountDownLatch;
public class CountDownLatch_Demo {
public static void main(String[] args) throws InterruptedException {
MyCountDownLatch latch = new MyCountDownLatch(6); //计数为6
for (int i = 0; i <6 ; i++) {
new Thread(()->{
System.out.println("开始准备.....");
latch.countDown();//计数减一
}).start();
Thread.sleep(1000);
}
latch.await(); //每个线程执行一次，则-1，在latch为0的时候开始向下运行 这是这些线程都准备就绪，然后去一起干同一件事
//还有一种方式， 将一个活分为多段，每个线程去干一段
// for (int i = 0; i <6 ; i++) {
// new Thread(()->{
// latch.countDown(); // 计数减一
// try {
// latch.await(); // 阻塞 -- > 0
// System.out.println("线程："+Thread.currentThread().getName()+"执行完毕");
// } catch (InterruptedException e) {
// e.printStackTrace();
// }
// }).start();
// }
System.out.println("开始干活....");
}
}
执行结果
开始准备.....
开始准备.....
开始准备.....
开始准备.....
开始准备.....
开始准备.....
开始干活....
3. CyclicBarrier （依赖了可重入锁的Condition和 signalAll）
循环栅栏，可以循环利用的屏障。
举例:排队上摩天轮时，每到齐四个人，就可以上同一个车厢。
例子
package concurrent;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
public class CyclicBarrier_Demo {
public static void main(String[] args) throws InterruptedException {
CyclicBarrier barrier = new CyclicBarrier(4);
for (int i = 0; i < 100; i++) { //假设有100个任务，每次只能有固定数量的线程去执行，可以使用这个
new Thread(() -> {
try {
barrier.await();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (BrokenBarrierException e) {
e.printStackTrace();
}
System.out.println("任务开始执行");
}).start();
Thread.sleep(500L);
}
}
}
- 自己实现CyclicBarrier 的例子
package concurrent;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
public class MyCyclicBarrier {
private final ReentrantLock lock = new ReentrantLock();
private final Condition condition = lock.newCondition();
private int count = 0; //批次
private final int parties; //多少线程准备就绪？
private Object generation = new Object(); // 类似于版本号，解决伪唤醒的问题
public MyCyclicBarrier(int parties) {
this.parties = parties;
}
public void await() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
final Object g = generation;
int index = ++count;
if (index == parties) {
nextGeneration();
return;
}
while (true) {
try {
condition.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
if (g != generation) {
return;
}
}
} finally {
lock.unlock();
}
}
public void nextGeneration() {
condition.signalAll();
count = 0;
generation = new Object();
}
}
执行代码
package concurrent;
public class CyclicBarrier_Demo {
public static void main(String[] args) throws InterruptedException {
MyCyclicBarrier barrier = new MyCyclicBarrier(4);
for (int i = 0; i < 100; i++) { //假设有100个任务，每次只能有固定数量的线程去执行，可以使用这个
new Thread(() -> {
barrier.await();
System.out.println("任务开始执行");
}).start();
Thread.sleep(500L);
}
}
}
如果觉得有收获就点个赞吧，更多知识，请点击关注查看我的主页信息哦~
