如何自己建设淘宝网站,搜索引擎搜索,网站制作 佛山,网站空间域名购买前言
在Java中#xff0c;AbstractQueuedSynchronizer#xff08;简称AQS#xff09;是一个用于实现同步器的抽象类#xff0c;它为实现各种类型的同步器#xff08;如锁、信号量等#xff09;提供了基本的框架。AQS通过一个双向队列#xff08;等待队列#xff09;和…前言
在Java中AbstractQueuedSynchronizer简称AQS是一个用于实现同步器的抽象类它为实现各种类型的同步器如锁、信号量等提供了基本的框架。AQS通过一个双向队列等待队列和一个整数变量状态来管理线程的排队和状态控制。
AQS的具体实现类主要有以下几种
ReentrantLock可重入锁的实现类支持独占模式。ReentrantReadWriteLock可重入读写锁的实现类支持共享和独占模式。Semaphore信号量的实现类用于控制同时访问某个资源的线程数量。CountDownLatch倒计时门闩的实现类用于等待一组线程完成某个操作后再执行。CyclicBarrier循环屏障的实现类用于等待一组线程都到达某个状态点后再一起继续执行。Phaser分阶段屏障的实现类用于协调多个线程在不同阶段的同步。LockSupport用于创建锁和其他同步类的基本线程阻塞原语。
其中ReentrantLock和ReentrantReadWriteLock是比较常用的同步器实现类用于提供可重入的独占锁和读写锁。Semaphore、CountDownLatch、CyclicBarrier和Phaser等同步器类则在特定场景下发挥重要作用帮助线程协调、控制和同步操作。
这些AQS实现类可以根据具体的需求选择使用通过继承AbstractQueuedSynchronizer类并实现其中的抽象方法可以定制和扩展自定义的同步器。
CountDownLatch
简介
CountDownLatch是Java中的一个同步辅助类用于实现线程间的等待和通知机制。它通过一个计数器来实现该计数器被初始化为一个正整数并且只能递减。线程在等待阶段通过调用await()方法等待计数器变为0而其他线程在完成自己的任务后通过调用countDown()方法来减少计数器的值。当计数器的值变为0时所有等待的线程都将被唤醒继续执行。 CountDownLatch的主要方法包括
void await()当前线程等待计数器的值变为0。如果计数器的值大于0await()方法将导致线程阻塞直到计数器的值变为0或被中断。 void countDown()将计数器的值减1。每个调用countDown()方法的线程都会使计数器减少1。 long getCount()获取当前计数器的值。 CountDownLatch通常用于以下场景
启动多个线程等待某个任务的完成主线程在创建需要等待的线程后通过CountDownLatch的构造函数将计数器的值设置为等待的线程数。每个线程在完成任务后调用countDown()方法计数器的值减1。主线程在需要等待的位置调用await()方法直到计数器的值变为0才会继续执行。 并行任务的等待和合并多个线程同时执行某个任务但是要求它们在继续执行前等待其他所有线程都完成。每个线程在完成自己的任务后调用countDown()方法主线程调用await()方法等待所有线程完成。 测试并发性能可以使用CountDownLatch来同时启动多个线程然后在主线程中等待所有线程完成以测试并发操作的性能。 通过使用CountDownLatch我们可以实现线程之间的同步和协调确保某些线程在其他线程完成任务后再继续执行以及在需要等待多个线程完成后再进行下一步操作。
其中内部有一个类Sync可以看到该类继承AbstractQueuedSynchronizer拥有AQS特性。 从构造函数来看Sync(int count)构造时传递一个线程状态存储。 //CountDownLatch的同步控制。使用AQS状态表示计数。private static final class Sync extends AbstractQueuedSynchronizer {private static final long serialVersionUID 4982264981922014374L;Sync(int count) {setState(count);}int getCount() {return getState();}protected int tryAcquireShared(int acquires) {return (getState() 0) ? 1 : -1;}protected boolean tryReleaseShared(int releases) {// Decrement count; signal when transition to zerofor (;;) {int c getState();if (c 0)return false;int nextc c-1;if (compareAndSetState(c, nextc))return nextc 0;}}}函数
await await()函数是CountDownLatch的核心函数之一用于使当前线程在计数器倒计数至零之前一直等待。如果线程被中断则会抛出InterruptedException。 函数内部通过调用sync.acquireSharedInterruptibly(1)将操作转发给了Sync对象的acquireSharedInterruptibly方法。
public void await() throws InterruptedException {// 转发到sync对象上sync.acquireSharedInterruptibly(1);
}acquireSharedInterruptibly
acquireSharedInterruptibly(int arg)方法在AQS中定义用于在获取共享资源时可中断地等待。如果线程被中断则会抛出InterruptedException。
acquireSharedInterruptibly(int arg)方法内部先检查线程的中断状态如果被中断则抛出InterruptedException。
public final void acquireSharedInterruptibly(int arg)throws InterruptedException {if (Thread.interrupted())throw new InterruptedException();if (tryAcquireShared(arg) 0)doAcquireSharedInterruptibly(arg);
}tryAcquireShared
如果通过tryAcquireShared(arg)方法尝试获取共享资源失败返回值小于0则调用doAcquireSharedInterruptibly(arg)方法进行进一步的等待操作。 tryAcquireShared(int acquires)是Sync类的方法它根据AQS的状态来判断是否可以获取共享资源。在CountDownLatch的实现中该方法简单地判断AQS的状态是否为0如果为0则返回1表示可以获取共享资源否则返回-1表示不能获取共享资源
protected int tryAcquireShared(int acquires) {return (getState() 0) ? 1 : -1;
}示例1
创建了一个CountDownLatch对象并将其初始计数值设置为3numOfThreads。然后我们创建了3个工作线程Worker类的实例每个工作线程模拟执行任务的时间。每个工作线程完成任务后调用countDown()方法来减少计数器的值。 **主线程通过调用latch.await()来等待所有工作线程完成。**当计数器的值变为0时主线程被唤醒并打印出All workers have completed their tasks.的消息。 该示例演示了如何使用CountDownLatch实现主线程等待多个工作线程完成任务后再继续执行的场景。
import java.util.concurrent.CountDownLatch;public class CountDownLatchExample {public static void main(String[] args) {int numOfThreads 3;CountDownLatch latch new CountDownLatch(numOfThreads);for (int i 0; i numOfThreads; i) {Thread thread new Thread(new Worker(latch));thread.start();}try {latch.await(); // 主线程等待所有工作线程完成System.out.println(All workers have completed their tasks.);} catch (InterruptedException e) {e.printStackTrace();}}static class Worker implements Runnable {private final CountDownLatch latch;public Worker(CountDownLatch latch) {this.latch latch;}Overridepublic void run() {// 模拟每个工作线程执行任务的时间try {Thread.sleep(2000);System.out.println(Worker completed its task.);} catch (InterruptedException e) {e.printStackTrace();} finally {latch.countDown(); // 每个工作线程完成任务后调用countDown()}}}
}} 在上面的示例中我们创建了一个CountDownLatch对象并将其初始计数值设置为3numOfThreads。然后我们创建了3个工作线程Worker类的实例每个工作线程模拟执行任务的时间。每个工作线程完成任务后调用countDown()方法来减少计数器的值。 主线程通过调用latch.await()来等待所有工作线程完成。当计数器的值变为0时主线程被唤醒并打印出All workers have completed their tasks.的消息。 该示例演示了如何使用CountDownLatch实现主线程等待多个工作线程完成任务后再继续执行的场景。
示例2
业务场景某个业务操作非常耗时但又必须等这个操作结束后才能进行后续操作。 import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.stream.Collectors;
import java.util.stream.Stream;/*** 多线程任务处理工具类*/
public class TaskDisposeUtils {//并行线程数public static final int POOL_SIZE;static {//判断核心线程数 如果机器的核心线程数大于5则用机器核心线程数POOL_SIZE Integer.max(Runtime.getRuntime().availableProcessors(), 5);}/*** 并行处理并等待结束**/public static T void dispose(ListT taskList, ConsumerT consumer) throws InterruptedException {dispose(true, POOL_SIZE, taskList, consumer);}/*** 并行处理并等待结束**/public static T void dispose(boolean moreThread, int poolSize, ListT taskList, ConsumerT consumer) throws InterruptedException {if (CollectionUtils.isEmpty(taskList)) {return;}//如果是多线程且核心线程数大于一则进入方法if (moreThread poolSize 1) {poolSize Math.min(poolSize, taskList.size());ExecutorService executorService null;try {//新建一个固定大小的线程池 核心线程数为poolSizeexecutorService Executors.newFixedThreadPool(poolSize);//juc工具类 用于让必须所有任务都处理完后才进行下一步CountDownLatch countDownLatch new CountDownLatch(taskList.size());for (T item : taskList) {executorService.execute(() - {try {//消费任务consumer.accept(item);} finally {//处理完后减一countDownLatch.countDown();}});}//在此等待 当countDownLatch变成0后才继续进行下一步countDownLatch.await();} finally {if (executorService ! null) {executorService.shutdown();}}} else {for (T item : taskList) {consumer.accept(item);}}}public static void main(String[] args) throws InterruptedException {//生成1-10的10个数字放在list中相当于10个任务ListInteger list Stream.iterate(1, a - a 1).limit(10).collect(Collectors.toList());JSONObject objectnew JSONObject();object.put(name,sss);//启动多线程处理list中的数据每个任务休眠时间为list中的数值
// ConsumerInteger c item - {
// try {
// long startTime System.currentTimeMillis();
// object.put(s,item);
// TimeUnit.SECONDS.sleep(item);
// long endTime System.currentTimeMillis();
// System.out.println(object.toJSONString());
// System.out.println(System.currentTimeMillis() ,任务 item 执行完毕耗时: (endTime - startTime));
// } catch (InterruptedException e) {
// e.printStackTrace();
// }
// };TaskDisposeUtils.dispose(list, item - {try {long startTime System.currentTimeMillis();object.put(s,item);TimeUnit.SECONDS.sleep(item);long endTime System.currentTimeMillis();System.out.println(object.toJSONString());System.out.println(System.currentTimeMillis() ,任务 item 执行完毕耗时: (endTime - startTime));} catch (InterruptedException e) {e.printStackTrace();}});//上面所有任务处理完毕完毕之后程序才能继续System.out.println(list 中的任务都处理完毕!);}
}
