• Java - AQS-CountDownLatch实现类（二）

    前言

    在Java中，AbstractQueuedSynchronizer（简称AQS）是一个用于实现同步器的抽象类，它为实现各种类型的同步器（如锁、信号量等）提供了基本的框架。AQS通过一个双向队列（等待队列）和一个整数变量（状态）来管理线程的排队和状态控制。

    AQS的具体实现类主要有以下几种：

    1. ReentrantLock：可重入锁的实现类，支持独占模式。
    2. ReentrantReadWriteLock：可重入读写锁的实现类，支持共享和独占模式。
    3. Semaphore：信号量的实现类，用于控制同时访问某个资源的线程数量。
    4. CountDownLatch：倒计时门闩的实现类，用于等待一组线程完成某个操作后再执行。
    5. CyclicBarrier：循环屏障的实现类，用于等待一组线程都到达某个状态点后再一起继续执行。
    6. Phaser：分阶段屏障的实现类，用于协调多个线程在不同阶段的同步。
    7. LockSupport：用于创建锁和其他同步类的基本线程阻塞原语。

    其中，ReentrantLock和ReentrantReadWriteLock是比较常用的同步器实现类，用于提供可重入的独占锁和读写锁。Semaphore、CountDownLatch、CyclicBarrier和Phaser等同步器类则在特定场景下发挥重要作用，帮助线程协调、控制和同步操作。

    这些AQS实现类可以根据具体的需求选择使用，通过继承AbstractQueuedSynchronizer类并实现其中的抽象方法，可以定制和扩展自定义的同步器。

    CountDownLatch

    简介

    CountDownLatch是Java中的一个同步辅助类，用于实现线程间的等待和通知机制。它通过一个计数器来实现，该计数器被初始化为一个正整数，并且只能递减。线程在等待阶段通过调用await()方法等待计数器变为0，而其他线程在完成自己的任务后通过调用countDown()方法来减少计数器的值。当计数器的值变为0时，所有等待的线程都将被唤醒继续执行。
    CountDownLatch的主要方法包括：

    void await()：当前线程等待计数器的值变为0。如果计数器的值大于0，await()方法将导致线程阻塞，直到计数器的值变为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 zero
            for (;;) {
                int c = getState();
                if (c == 0)
                    return false;
                int nextc = c-1;
                if (compareAndSetState(c, nextc))
                    return nextc == 0;
            }
        }
    }

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28

    函数

    await

    • await()函数是CountDownLatch的核心函数之一，用于使当前线程在计数器倒计数至零之前一直等待。如果线程被中断，则会抛出InterruptedException。

    • 函数内部通过调用sync.acquireSharedInterruptibly(1)将操作转发给了Sync对象的acquireSharedInterruptibly方法。

    public void await() throws InterruptedException {
    // 转发到sync对象上
    sync.acquireSharedInterruptibly(1);
}

    • 1
    • 2
    • 3
    • 4
    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);
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    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
    • 2
    • 3

    示例1

    创建了一个CountDownLatch对象，并将其初始计数值设置为3（numOfThreads）。然后，我们创建了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;
        }

        @Override
        public void run() {
            // 模拟每个工作线程执行任务的时间
            try {
                Thread.sleep(2000);
                System.out.println("Worker completed its task.");
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                latch.countDown(); // 每个工作线程完成任务后调用countDown()
            }
        }
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41

    }
    在上面的示例中，我们创建了一个CountDownLatch对象，并将其初始计数值设置为3（numOfThreads）。然后，我们创建了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(List<T> taskList, Consumer<T> consumer) throws InterruptedException {
        dispose(true, POOL_SIZE, taskList, consumer);
    }
    /**
     * 并行处理，并等待结束
     *
     */
    public static <T> void dispose(boolean moreThread, int poolSize, List<T> taskList, Consumer<T> consumer) throws InterruptedException {
        if (CollectionUtils.isEmpty(taskList)) {
            return;
        }
        //如果是多线程且核心线程数大于一则进入方法
        if (moreThread && poolSize > 1) {
            poolSize = Math.min(poolSize, taskList.size());
            ExecutorService executorService = null;
            try {
                //新建一个固定大小的线程池 核心线程数为poolSize
                executorService = 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个任务
        List<Integer> list = Stream.iterate(1, a -> a + 1).limit(10).collect(Collectors.toList());
        JSONObject object=new JSONObject();
        object.put("name","sss");
        //启动多线程处理list中的数据，每个任务休眠时间为list中的数值
//        Consumer 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 + "中的任务都处理完毕!");
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76
    • 77
    • 78
    • 79
    • 80
    • 81
    • 82
    • 83
    • 84
    • 85
    • 86
    • 87
    • 88
    • 89
    • 90
    • 91
    • 92
    • 93
    • 94
    • 95
    • 96
    • 97
    • 98
    • 99
    • 100
    • 101
    • 102
  • 相关阅读:
    前端无需install快速调试npm包，Console-Import使用
    Mybatis 快速入门之 动态sql和分页
    Kubernetes(k8s)网络策略NetworkPolicy
    python,修改数组【第十届】【省赛】【研究生组】
    MySQL中支持的字符集和排序规则
    一名GISer的本科生涯--写于2018年
    力扣--76. 最小覆盖子串
    重学设计模式（三、设计模式-备忘录模式）
    哪些浏览器受到用户欢迎？分享这两款安全浏览器
    MybatisPlus【SpringBoot】 6 插件 6.1 分页插件 & 6.2 xml 自定义分页
  • 原文地址：https://blog.csdn.net/qq_42583549/article/details/130915748