并发编程——JUC并发工具

2023-09-21 17:07:28


在这里插入图片描述

前言

JUC 是Java并发编程工具类库,提供了一些常用的并发工具,例如锁、信号量、计数器、事件循环、线程池、并发集合等。这些工具可以帮助开发人员简化并发编程的复杂性,提高程序效率和可靠性。

CountDownLatch

CountDownLatch应用

CountDownLatch本身就好像一个计数器,可以等待一个或多个线程完成后再执行,其基于AQS实现。

public static void main(String[] args) throws InterruptedException, BrokenBarrierException {
    CountDownLatch countDownLatch = new CountDownLatch(3);

    new Thread(() -> {
        System.out.println("111");
        countDownLatch.countDown();
    }).start();

    new Thread(() -> {
        System.out.println("222");
        countDownLatch.countDown();
    }).start();

    new Thread(() -> {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("333");
        countDownLatch.countDown();
    }).start();

    // 主线会阻塞在这个位置,直到CountDownLatch的state变为0
    countDownLatch.await();
    System.out.println("main");
}

CountDownLatch核心源码

// CountDownLatch 的有参构造
public CountDownLatch(int count) {
    // 健壮性校验
    if (count < 0) throw new IllegalArgumentException("count < 0");
    // 构建Sync给AQS的state赋值
    this.sync = new Sync(count);
}

countDown方法,本质就是调用了AQS的释放共享锁操作

public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) {
        // 唤醒在AQS队列中排队的线程。
        doReleaseShared();
        return true;
    }
    return false;
}

// countDownLatch实现的业务
protected boolean tryReleaseShared(int releases) {
    for (;;) {
        int c = getState();
        if (c == 0)
            return false;
        // state - 1
        int nextc = c-1;
        // 用CAS赋值
        if (compareAndSetState(c, nextc))
            return nextc == 0;
    }
}
// 如果CountDownLatch中的state已经为0了,那么再次执行countDown跟没执行一样。
// 而且只要state变为0,await就不会阻塞线程。

功能都是AQS提供的,只有tryReleaseShared需要实现的类自己去编写业务。

await方法,调用了AQS提供的获取共享锁并且允许中断的方法

// await方法
public void await() throws InterruptedException {
    sync.acquireSharedInterruptibly(1);
}

// AQS获取共享锁并且允许中断的方法
public final void acquireSharedInterruptibly(int arg)
        throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    // countDownLatch操作
    if (tryAcquireShared(arg) < 0)
        // 如果返回的是-1,代表state肯定大于0
        doAcquireSharedInterruptibly(arg);
}

// CountDownLatch实现的tryAcquireShared
protected int tryAcquireShared(int acquires) {
    // state为0,返回1,。否则返回-1
    return (getState() == 0) ? 1 : -1;
}

// 让当前线程进到AQS队列,排队去
private void doAcquireSharedInterruptibly(int arg) throws InterruptedException {
    // 将当前线程封装为Node,并且添加到AQS的队列中
    final Node node = addWaiter(Node.SHARED);
    boolean failed = true;
    try {
        for (;;) {
            final Node p = node.predecessor();
            if (p == head) {
                // 再次走上面的tryAcquireShared,如果返回的是的1,代表state为0
                int r = tryAcquireShared(arg);
                if (r >= 0) {
                    // 会将当前线程和后面所有排队的线程都唤醒。
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    failed = false;
                    return;
                }
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                throw new InterruptedException();
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}


Semaphore

Semaphore应用

Semaphore一般用于流控。比如有一个公共资源,多线程都可以访问时,Semaphore可以当作信号量做限制。每当有一个线程获取连接对象时,对信号量-1,当这个线程归还资源时对信号量+1。如果线程拿资源时,发现Semaphore内部的资源个数为0,就会被阻塞。

public static void main(String[] args) throws InterruptedException, BrokenBarrierException {
    // 声明信号量
    Semaphore semaphore = new Semaphore(1);
    // 能否去拿资源
    semaphore.acquire();
    // 拿资源处理业务
    System.out.println("main");
    // 归还资源
    semaphore.release();
}

Semaphore核心源码

Semaphore有公平和非公平两种竞争资源的方式。

// 
public Semaphore(int permits, boolean fair) {
    sync = fair ? new FairSync(permits) : new NonfairSync(permits);
}

// 设置资源个数,State其实就是信号量的资源个数
Sync(int permits) {
    setState(permits);
}

在调用 acquire 获取资源时也是基于AQS提供的获取共享锁方法。

release就是将state+1,归还资源。

// 两个一起 阿巴阿巴
public void release() {
    sync.releaseShared(1);
}

public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) {
        // 唤醒在AQS中排队的Node,去竞争资源
        doReleaseShared();
        return true;
    }
    return false;
}

// 信号量实现的归还资源
protected final boolean tryReleaseShared(int releases) {
    for (;;) {
        // 拿state
        int current = getState();
        // state + 1
        int next = current + releases;
        // 资源最大值,再+1,变为负数
        if (next < current)
            throw new Error("Maximum permit count exceeded");
        // CAS 改一手
        if (compareAndSetState(current, next))
            return true;
    }
}

共享锁在释放资源后,如果头节点为0,无法确认真的没有后继节点。如果头节点为0,需要将头节点的状态修改为-3,当最新拿到锁资源的线程,查看是否有后继节点并且为共享锁,就唤醒排队的线程

CyclicBarrier

CyclicBarrier应用

CyclicBarrier 一般称为栅栏,和CountDownLatch很像。CountDownLatch在操作时,只能使用一次,也就是state变为0之后,就无法再使用了。CyclicBarrier是可以复用的,他的计数器可以归位,然后再处理。而且可以在计数过程中出现问题后,重置当前CyclicBarrier,再次重新操作!

public static void main(String[] args) throws InterruptedException, BrokenBarrierException {
    // 声明栅栏
    CyclicBarrier barrier = new CyclicBarrier(3,() -> {
        System.out.println("开始!");
    });

    new Thread(() -> {
        System.out.println("第一位选手到位");
        try {
            barrier.await();
            System.out.println("第一位往死里跑!");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }).start();

    new Thread(() -> {
        System.out.println("第二位选手到位");
        try {
            barrier.await();
            System.out.println("第二位也往死里跑!");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }).start();

    System.out.println("裁判已经到位");
    barrier.await();
}

CyclicBarrier核心源码

CyclicBarrier没有直接使用AQS,而是使用ReentrantLock,间接的使用AQS

// CyclicBarrier的有参
public CyclicBarrier(int parties, Runnable barrierAction) {// 健壮性判断!
    if (parties <= 0) throw new IllegalArgumentException();
    // parties是final修饰的,需要在重置时,使用!
    this.parties = parties;
    // count是在执行await用来计数的。
    this.count = parties;
    // 当计数count为0时 ,先执行这个Runnnable!在唤醒被阻塞的线程
    this.barrierCommand = barrierAction;
}

线程执行await方法,会对count-1,再判断count是否为0,如果不为0,需要添加到AQS中的ConditionObject的Waiter队列中排队,并park当前线程。如果为0,证明线程到齐,需要执行nextGeneration,会先将Waiter队列中的Node全部转移到AQS的队列中,没有后继节点设置为0。然后重置count和broker标记。等到unlock执行后,每个线程都会被唤醒。

private int dowait(boolean timed, long nanos) throws InterruptedException, BrokenBarrierException, TimeoutException {
    // 相当于synchronized中使用wait和notify
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        // 里面就是boolean,默认false
        final Generation g = generation;

        // 判断之前栅栏加入线程时,是否有超时、中断等问题,如果有,设置boolean为true,其他线程再进来,直接凉凉
        if (g.broken)
            throw new BrokenBarrierException();

        if (Thread.interrupted()) {
            breakBarrier();
            throw new InterruptedException();
        }


        // 对计数器count--
        int index = --count;
        // 如果--完,是0,代表突破栅栏,干活!
        if (index == 0) {  
            // 默认false
            boolean ranAction = false;
            try {
                // 如果你用的是2个参数的有参构造,说明你传入了任务,index == 0,先执行CyclicBarrier有参的任务
                final Runnable command = barrierCommand;
                if (command != null)
                    command.run();
                // 设置为true
                ranAction = true;
                nextGeneration();
                return 0;
            } finally {
                if (!ranAction)
                    breakBarrier();
            }
        }

        // --完之后,index不是0,代表还需要等待其他线程
        for (;;) {
            try {
                // 如果没设置超时时间。  await()
                if (!timed)
                    trip.await();
                // 设置了超时时间。  await(1,SECOND)
                else if (nanos > 0L)
                    nanos = trip.awaitNanos(nanos);
            } catch (InterruptedException ie) {
                if (g == generation && ! g.broken) {
                    breakBarrier();
                    throw ie;
                } else {
                    Thread.currentThread().interrupt();
                }
            }

            if (g.broken)
                throw new BrokenBarrierException();

            if (g != generation)
                return index;

            if (timed && nanos <= 0L) {
                breakBarrier();
                throw new TimeoutException();
            }
        }
    } finally {
        lock.unlock();
    }
}



// 挂起线程
public final void await() throws InterruptedException {
    // 允许中断
    if (Thread.interrupted())
        throw new InterruptedException();
    // 添加到队列(不是AQS队列,是AQS里的ConditionObject中的队列)
    Node node = addConditionWaiter();
    int savedState = fullyRelease(node);
    int interruptMode = 0;
    while (!isOnSyncQueue(node)) {
        // 挂起当前线程
        LockSupport.park(this);
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
            break;
    }
}


// count到0,唤醒所有队列里的线程线程
private void nextGeneration() {
    // 这个方法就是将Waiter队列中的节点遍历都扔到AQS的队列中,真正唤醒的时机,是unlock方法
    trip.signalAll();
    // 重置计数器
    count = parties;
    // 重置异常判断
    generation = new Generation();
}

总结

使用这些工具类时需要注意:

  • Semaphore的使用要避免死锁和过度同步导致的性能问题。
  • CyclicBarrier在屏障点之后的代码要保证所有线程都能正确执行,否则可能导致部分线程一直等待。
  • CountDownLatch的countDown方法要保证在所有线程执行完毕之前被调用,否则可能导致部分线程一直等待。

根据具体的应用场景选择合适的工具类,正确使用并合理设计并发策略,可以提高程序的效率和可靠性。

更多推荐

zabbix自定义监控、钉钉、邮箱报警 (五十六)

提示:文章写完后,目录可以自动生成,如何生成可参考右边的帮助文档目录一、实验准备二、安装三、添加监控对象四、添加自定义监控项五、监控mariadb1、添加模版查看要求2、安装mariadb、创建用户3、创建用户文件4、修改监控模版5、在上述文件中配置路径6、重启zabbix-agent验证六、监控NGINX1、安装NG

干货!网络丢包故障定位全景指南

极客星球:深入理解网络子系统欢迎加入极客星球,一起学习,快速成长:修炼基本功,打好高薪基础:分享基础技术深度理解,基础概念深度解析,经典书籍推荐和读书分享,经典源码阅读分享等,不定期直播分享,问题答疑解惑;扩展技术深度和技术视野:分享热门技术发展,国内外大厂技术内幕,业界解决方案;校招/社招免费就业指导:模拟面试,简历

精准测试(针对人工执行的测试用例和自动化测试脚本)

在软件测试中,我们常常碰到两个基本问题(困难):很难保障无漏测:我们做了大量测试,但不清楚测得怎样,对软件上线后会不会出问题,没有信心;选择待执行的测试用例:面对大量的回归测试用例时,我们没有足够的时间完成测试,如何选择出有效的测试用例呢?虽然我们会有一些策略,如基于风险的测试策略、基于操作剖面的测试策略或组合测试策略

系统架构:软件工程速成

文章目录参考概述软件工程概述软件过程可行性分析可行性分析概述数据流图数据字典需求分析需求分析概述ER图状态转换图参考软件工程速成(期末+考研复试+软考)均适用.支持4K概述软件工程概述定义:采用工程的概念、原理、技术和方法来开发与维护软件。三要素:方法:完成软件开发各项任务的技术方法,回答“怎么做”。工具:为运用方法提

美创科技入选IDC中国等保合规市场报告推荐厂商

近日,全球领先的IT研究和咨询公司IDC正式发布《IDCPersepctive:中国等保合规市场洞察,2023》报告。在该市场报告中,IDC对中国等保合规市场的发展现状进行调研,明确了最终用户等保合规建设的痛点、难点,阐述了市场中各技术服务提供商的产品服务方案和优势。美创科技凭借在数据安全和运行安全领域专业能力与积累实

垃圾收集算法

1.如何判断对象是否存活?1.1引用计数算法基本思路:在对象中添加一个引用计数器每当有一个地方引用它的时候,计数器就加+1每当有一个引用失效的时候,计数器就减-1当计数器的值为0的时候,那么该对象就是可被GC回收的垃圾对象存在的问题:对象循环引用a对象引用了b对象,b对象也引用了a对象,a、b对象却没有再被其他对象所引

单片机外设-串口(UART)详情

目录学习UART要先认识一些基础知识一:什么是串行、并行通信?(1)串行通信串行通信概念:串行通信的特点:(2)并行通信并行通信概念:并行通信特点:二:什么是异步通信、同步通信?(1)异步通信​编辑异步通信概念:异步通信特点:(2)同步通信同步通信概念:1、外同步2、自同步同步通信特点:三:什么是单工、半双工、全双工通

Vue知识系列(5)每天10个小知识点

目录系列文章目录Vue知识系列(1)每天10个小知识点Vue知识系列(2)每天10个小知识点Vue知识系列(3)每天10个小知识点Vue知识系列(4)每天10个小知识点知识点41.vue常用基本指令有哪些以及他们的作用和使用场景42.Vue组件中data为什么必须是函数43.v-if和v-show的区别44.vue自定

Git(7)——使用Beyond Compare快速解决冲突

一、简介根据前六章的学习,我们应该很清楚地感知到不同分支合并代码时产生的冲突是最让我们头疼的问题,因为他需要我们手动去解决冲突的文件,有没有一种方法可以快速地解决冲突呢?本篇文章将介绍如何使用ByondCompare去快速解决冲突二、在Git中进行配置使用如下命令对Git进行配置注:这里的--local是指以下这命令配

Nacos内核设计之一致性协议(上)

Nacos一致性协议Nacos技术架构先简单介绍下Nacos的技术架构从而对nacos有一个整体的认识如图Nacos架构分为四层用户层、应用层、核心层、各种插件再深入分析下nacos一致性协议的发展过程及原理实现为什么nacos需要一致性协议Nacos是一个需要存储数据的一个组件为了实现这个目标,就需要在Nacos内部

(2022,DALL·E2,CLIP,Diffusion,AR)使用 CLIP 潜在空间的分层文本条件图像生成

HierarchicalText-ConditionalImageGenerationwithCLIPLatents公众号:EDPJ(添加VX:CV_EDPJ或直接进Q交流群:922230617获取资料)目录0.摘要1.简介2.方法2.1解码器2.2先验3.图像处理3.1变化3.2插值3.3文本差异(TextDiffs

热文推荐