Tim

一枚野生程序员~

  • 主页
  • 分类
  • 标签
  • 归档
  • 关于
所有文章 工具

Tim

一枚野生程序员~

  • 主页
  • 分类
  • 标签
  • 归档
  • 关于

JUC的四个并发工具类

阅读数:次 2020-04-05
字数统计: 1.6k字   |   阅读时长≈ 6分

本篇文章主要记录了JUC的四个并发工具类,闭锁CountDownlatch、栅栏CyclicBarrier、信号量Semaphore、交换器Exchanger。CountDownlatch通常用于主线程等待其他任务线程执行完毕的场景;CyclicBarrier主要阻塞当前线程,等待其他线程(大家无论谁先跑到A点,必须要等其他线程也到达了A点,大家才能继续)。信号量Semaphore可以用来控制同时访问特定资源的线程数量(比如100个线程只能有10个线程可以获得MySQL连接)。交换器Exchanger很少用,只适用于两个线程在同步点交换数据的场景(如下图)。

mark

闭锁CountDownlatch

CountDownLatch也叫闭锁,使得一(多)个主线程必须等待其他线程完成操作后再执行。 CountDownLatch内部维护一个计数器(父类的int state),主线程先执行await方法,如果此时计数器大于0,则阻塞等待。当一个线程完成任务后,计数器值减1。直到计数器为0时,表示所有的线程已经完成任务,等待的主线程被唤醒继续执行。

CountDownLatch实现主要基于Java同步器AQS,关于AQS的分析可以看我这篇文章:《ReentrantLock与AQS》 ,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
public void countDown() {
sync.releaseShared(1);
}

public final boolean releaseShared(int arg) {
// countDownLatch -> tryReleaseShared
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}

private void doReleaseShared() {
for (;;) {
Node h = head;
if (h != null && h != tail) {
int ws = h.waitStatus;
if (ws == Node.SIGNAL) {
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue; // loop to recheck cases
unparkSuccessor(h);
}
else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue; // loop on failed CAS
}
if (h == head) // loop if head changed
break;
}
}

CountDownlatch的使用案例,下面使用三个线程来打印三个List,三个线程任务都完成得时候才输出Print Task Finish!

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
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CountDownLatch;

public class CountDownLatchDemo {
public static void main(String[] args) {
List<String> list1 = Arrays.asList("AAA", "BBB", "CCC");
List<String> list2 = Arrays.asList("DDD", "EEE", "FFF");
List<String> list3 = Arrays.asList("GGG", "HHH", "III");

CountDownLatch countDownLatch = new CountDownLatch(3);
new Thread(()->{
for (String string: list1) {
System.out.println(Thread.currentThread().getName() + ":" + string);
}
countDownLatch.countDown();
}).start();
new Thread(()->{
for (String string: list2) {
System.out.println(Thread.currentThread().getName() + ":" + string);
}
countDownLatch.countDown();
}).start();
new Thread(()->{
for (String string: list3) {
System.out.println(Thread.currentThread().getName() + ":" + string);
}
countDownLatch.countDown();
}).start();
try {
countDownLatch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("Print Task Finish!");
}
}

mark

栅栏CyclicBarrier

CyclicBarrier:阻塞当前线程,等待其他线程。等待其它线程,且会阻塞自己当前线程,所有线程必须同时到达栅栏位置后才能继续执行;所有线程到达栅栏处,可以触发执行另外一个预先设置的线程。

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
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;

public class CyclicBarrierDemo {
public static void main(String[] args) throws InterruptedException {
new CyclicBarrierDemo().go();
}

private void go() throws InterruptedException {
//初始化栅栏得参与者数为3
CyclicBarrier cyclicBarrier = new CyclicBarrier(3);
new Thread(new Task(cyclicBarrier), "Thread1").start();
Thread.sleep(1000);
new Thread(new Task(cyclicBarrier), "Thread2").start();
Thread.sleep(1000);
new Thread(new Task(cyclicBarrier), "Thread3").start();
}

class Task implements Runnable{
private CyclicBarrier cyclicBarrier;

public Task(CyclicBarrier cyclicBarrier) {
this.cyclicBarrier = cyclicBarrier;
}

@Override
public void run() {
System.out.println("线程" + Thread.currentThread().getName()
+ "已经送达"
+ System.currentTimeMillis());
try {
cyclicBarrier.await();
} catch (BrokenBarrierException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("线程" + Thread.currentThread().getName()
+ "开始处理"
+ System.currentTimeMillis());
}
}
}

mark

信号量Semaphore

Semaphore也叫信号量,在JDK1.5被引入,可以用来控制同时访问特定资源的线程数量,通过协调各个线程,以保证合理的使用资源。Semaphore内部维护了一组虚拟的许可,许可的数量可以通过构造函数的参数指定。

访问特定资源前,必须使用acquire方法获得许可,如果许可数量为0,该线程则一直阻塞,直到有可用许可。访问资源后,使用release释放许可。Semaphore和ReentrantLock类似,获取许可有公平策略和非公平许可策略,默认情况下使用非公平策略。

信号量Semaphore得应用场景:Semaphore可以用来做流量分流,特别是对公共资源有限的场景,比如数据库连接。假设有这个的需求,读取几万个文件的数据到数据库中,由于文件读取是IO密集型任务,可以启动几十个线程并发读取,但是数据库连接数只有10个,这时就必须控制最多只有10个线程能够拿到数据库连接进行操作。这个时候,就可以使用Semaphore做流量控制。

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
package thread_study;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;

public class SemaphoreDemo {
public static void main(String[] args) {
ExecutorService pool = Executors.newCachedThreadPool();
//只能5个线程同时访问
Semaphore semaphore = new Semaphore(5);
//模拟20个客户端访问
for (int i = 0; i < 20; i++) {
final int NO = i;
pool.execute(()->{
try {
//获取许可
semaphore.acquire();
System.out.println("Accessing: " + NO);
Thread.sleep((long)(Math.random() * 10000));
//访问完毕后释放
semaphore.release();
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
//退出线程池
pool.shutdown();
}
}

mark

交换器Exchanger

Exchanger(交换者)是一个用于线程间数据交换协作的工具类。它提供一个同步点,在这个同步点多个线程间两两之间线程可以交换彼此的数据。这两个线程通过exchange方法交换数据, 如果第一个线程先执行exchange方法,它会一直等待第二个线程也执行exchange方法,当两个线程都到达同步点时,这两个线程就可以交换数据,将本线程生产出来的数据传递给对方。

mark

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
import java.util.concurrent.Exchanger;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class ExchangerDemo {
public static void main(String[] args) {
Exchanger<String> exchanger = new Exchanger<>();
ExecutorService threadPool = Executors.newFixedThreadPool(2);
threadPool.execute(()->{
try {
//男生对女生说的话
String girl = exchanger.exchange("我其实暗恋你很久了....");
System.out.println("女生说: " + girl);
} catch (InterruptedException e) {
e.printStackTrace();
}
});

threadPool.execute(()->{
try {
System.out.println("女生慢慢的从教室走出来.... ");
TimeUnit.SECONDS.sleep(3);
//男生对女生说的话
String boy = exchanger.exchange("我很喜欢你....");
System.out.println("男生说:" + boy);
} catch (InterruptedException e) {
e.printStackTrace();
}
});

threadPool.shutdown();
}
}

mark

赏

谢谢你请我喝咖啡

支付宝
微信
  • 本文作者: Tim
  • 本文链接: https://zouchanglin.cn/2876604782.html
  • 版权声明: 本博客所有文章除特别声明外,均采用 CC BY-SA 4.0 许可协议。转载请注明出处!
  • JavaSE
  • JUC
  • 异步
  • 并发编程

扫一扫,分享到微信

如何理解SpringIOC
ForkJoin框架与读写锁
  1. 1. 闭锁CountDownlatch
  2. 2. 栅栏CyclicBarrier
  3. 3. 信号量Semaphore
  4. 4. 交换器Exchanger
© 2017-2021 Tim
本站总访问量次 | 本站访客数人
  • 所有文章
  • 工具

tag:

  • 生活
  • Android
  • 索引
  • MySQL
  • 组件通信
  • Nginx
  • JavaSE
  • JUC
  • JavaWeb
  • 模板引擎
  • 前端
  • Linux
  • 计算机网络
  • Docker
  • C/C++
  • JVM
  • 上传下载
  • JavaEE
  • SpringCloud
  • Golang
  • Gradle
  • 网络安全
  • 非对称加密
  • IDEA
  • SpringBoot
  • Jenkins
  • 字符串
  • vim
  • 存储
  • 文件下载
  • Mac
  • Windows
  • NIO
  • RPC
  • 集群
  • 微服务
  • SSH
  • 配置中心
  • XML
  • Chrome
  • 压力测试
  • Git
  • 博客
  • 概率论
  • 排序算法
  • 分布式
  • 异常处理
  • 文件系统
  • 哈希
  • openCV
  • 栈
  • 回溯
  • SpringCore
  • 流媒体
  • rtmp
  • 面向对象
  • Vue
  • ElementUI
  • 软件工程
  • 异步
  • 自定义UI
  • ORM框架
  • 模块化
  • 交互式
  • Jsoup
  • Http Client
  • LRUCache
  • RabbitMQ
  • 消息通信
  • 服务解耦
  • 负载均衡
  • 权限
  • 多线程
  • 单例模式
  • Protobuf
  • 序列化
  • Python
  • m3u8
  • 堆
  • 二叉树
  • 自定义View
  • 观察者模式
  • 设计模式
  • 线程池
  • 动态扩容
  • 高可用
  • GC
  • ffmpeg
  • SpringMVC
  • REST
  • Redis
  • 缓存中间件
  • UML
  • Maven
  • Netty
  • 高性能网络
  • IPC通信
  • IO
  • Stream
  • 发布订阅
  • SQLite
  • Hash
  • 集合框架
  • 链表
  • Lambda
  • 汇编语言
  • 组件化
  • Router
  • 开发工具

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录(注意不是yilia-plus根目录)执行以下命令:
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置:

      jsonContent:
        meta: false
        pages: false
        posts:
          title: true
          date: true
          path: true
          text: false
          raw: false
          content: false
          slug: false
          updated: false
          comments: false
          link: false
          permalink: false
          excerpt: false
          categories: false
          tags: true
    

  • 思维导图
  • PDF工具
  • 无损放大
  • 代码转图
  • HTTPS证书