JUC并发包基于AQS实现的线程同步器的案例分析

以下是JUC并发包提供的基于AQS实现的线程同步器。

ReentrantLock：可重入锁

通常用于多线程操作进行同步，实现线程安全。存在公平和非公平两种实现，默认为非公平。

如在LinkedBlockingQueue中使用putLock来同步多个生产者的写操作，使用takeLock来同步多个消费者的读操作。

public boolean offer(E e, long timeout, TimeUnit unit)throws InterruptedException {if (e == null) throw new NullPointerException();long nanos = unit.toNanos(timeout);int c = -1;final ReentrantLock putLock = this.putLock;final AtomicInteger count = this.count;// 获取写锁，成功则其他线程在这个地方等待putLock.lockInterruptibly();try {while (count.get() == capacity) {if (nanos <= 0)return false;nanos = notFull.awaitNanos(nanos);}enqueue(new Node<E>(e));c = count.getAndIncrement();if (c + 1 < capacity)notFull.signal();} finally {// 释放写锁，成功则其他线程（其中一个）可以从上面那里进行继续执行putLock.unlock();}if (c == 0)signalNotEmpty();return true;
}

CoutDownLatch：倒计时器

相当于一个倒计时器，需要在应用代码中来执行countDown递减该计时器。

等到指定数量的线程都完成了工作，则在主线程中执行汇总等操作，如一个复杂的计算分成多个小任务，在多个子线程里执行，最后在主线程来汇总；或者是实现控制N个子线程同时开始执行。

class Driver2 { // ...void main() throws InterruptedException {// 大任务拆成N个小任务CountDownLatch doneSignal = new CountDownLatch(N);Executor e = ...for (int i = 0; i < N; ++i) // create and start threadse.execute(new WorkerRunnable(doneSignal, i));doneSignal.await();           // wait for all to finish}
}class WorkerRunnable implements Runnable {private final CountDownLatch doneSignal;private final int i;WorkerRunnable(CountDownLatch doneSignal, int i) {this.doneSignal = doneSignal;this.i = i;}public void run() {try {doWork(i);doneSignal.countDown();} catch (InterruptedException ex) {} // return;}void doWork() { ... }
}

控制N个子线程同时开始执行

class Driver { // ...void main() throws InterruptedException {// 开始控制开关CountDownLatch startSignal = new CountDownLatch(1);// N个线程都准备就绪开关CountDownLatch doneSignal = new CountDownLatch(N);for (int i = 0; i < N; ++i) // create and start threadsnew Thread(new Worker(startSignal, doneSignal)).start();doSomethingElse();            // don't let run yet// 打开控制开关startSignal.countDown();      // let all threads proceeddoSomethingElse();// 等待N个线程都完成doneSignal.await();           // wait for all to finish}}class Worker implements Runnable {private final CountDownLatch startSignal;private final CountDownLatch doneSignal;Worker(CountDownLatch startSignal, CountDownLatch doneSignal) {this.startSignal = startSignal;this.doneSignal = doneSignal;}public void run() {try {// 等待主线程打开控制开关startSignal.await();doWork();// 该子线程完成了自身工作doneSignal.countDown();} catch (InterruptedException ex) {} // return;}void doWork() { ... }
}

CyclicBarric：栅栏

可以理解为一个栅栏，拦住线程直到所有线程到达则打开该栅栏让所有线程继续执行。
与CountDownLatch功能差不多，只是计数可重置为初始值，重复使用，具体为调用reset方法来实现，如果在执行过程中执行reset，则子线程会抛异常退出；同时可以指定一个Runnable任务在栅栏打开时来执行。由于可以重复使用，每次符合条件打开该栅栏时，都重复执行该Runnable任务。
内部也是使用ReentrantLock和Condition来实现的。
CyclicBarric使用all-or-none的执行模式，即要么所有成功，要么所有失败，任何一个子线程被中断或者异常，超时退出，则所有线程都退出。

以下例子为处理矩阵，每个线程处理矩阵的一行，当矩阵的所有行都处理完毕之后，使用barrierAction这个Runnable实例来执行汇总操作mergeRows。

class Solver {final int N;final float[][] data;final CyclicBarrier barrier;class Worker implements Runnable {int myRow;Worker(int row) { myRow = row; }public void run() {while (!done()) {processRow(myRow);try {// 等待其他线程处理完毕则返回，// 即N个线程都调用了barrier.awaitbarrier.await();} catch (InterruptedException ex) {return;} catch (BrokenBarrierException ex) {return;}}}}public Solver(float[][] matrix) {data = matrix;N = matrix.length;// 等栅栏打开，自动执行该任务Runnable barrierAction =new Runnable() { public void run() { mergeRows(...); }};// 新建栅栏，指定线程集合的大小barrier = new CyclicBarrier(N, barrierAction);List<Thread> threads = new ArrayList<Thread>(N);for (int i = 0; i < N; i++) {Thread thread = new Thread(new Worker(i));threads.add(thread);thread.start();}// wait until donefor (Thread thread : threads)thread.join();}
}

Semaphore：信号量

限制某种资源的可用数量，通常可以用于实现池化机制。也存在公平和非公平两种实现，默认为非公平。

class Pool {private static final int MAX_AVAILABLE = 100;private final Semaphore available = new Semaphore(MAX_AVAILABLE, true);public Object getItem() throws InterruptedException {available.acquire();return getNextAvailableItem();}public void putItem(Object x) {if (markAsUnused(x))available.release();}// Not a particularly efficient data structure; just for demoprotected Object[] items = ... whatever kinds of items being managedprotected boolean[] used = new boolean[MAX_AVAILABLE];protected synchronized Object getNextAvailableItem() {for (int i = 0; i < MAX_AVAILABLE; ++i) {if (!used[i]) {used[i] = true;return items[i];}}return null; // not reached}protected synchronized boolean markAsUnused(Object item) {for (int i = 0; i < MAX_AVAILABLE; ++i) {if (item == items[i]) {if (used[i]) {used[i] = false;return true;} elsereturn false;}}return false;}
}

JUC并发包基于AQS实现的线程同步器的案例分析相关推荐

Java多线程：线程8锁案例分析
线程8锁案例分析通过分析代码,推测打印结果,并运行代码进行验证 1.两个线程调用同一个对象的两个同步方法被synchronized修饰的方法,锁的对象是方法的调用者.因为两个方法的调用者是同一个, ...
非线性有限元：基本理论与算法及基于Python、Fortran程序实现与案例分析实践技术
有限单元法在岩土工程问题中应用非常广泛,很多商业软件如Plaxis/Abaqus/Comsol等都采用有限单元解法.尽管各类商业软件使用方便,但其使用对用户来说往往是一个"黑箱子" ...
非线性有限元：基本理论与算法及基于Python、Fortran程序实现与案例分析
非线性有限元:基本理论与算法及基于Python.Fortran程序实现与案例分析 (qq.com) 有限单元法在岩土工程问题中应用非常广泛,很多商业软件如Plaxis/Abaqus/Comsol等都采 ...
岩土工程--非线性有限元：基本理论与算法及基于Python、Fortran程序实现与案例分析
非线性有限元:基本理论与算法及基于Python.Fortran程序实现与案例分析实践技术有限单元法在岩土工程问题中应用非常广泛,很多商业软件如Plaxis/Abaqus/Comsol等都采用有限单元 ...
基于SOA的区域卫生信息平台案例分析(转)
这篇文章是在学习过程中baidud到得,觉得在学习阶段可以看看,用的是ESB做的系统集成. 以下是原文: 一.政策与现状国务院在<关于深化医药卫生体制改革意见>中明确提出要大力推进医药卫 ...
基于MATLAB的Dijkstra算法实现及案例分析
摘要:为研究两地点之间距离(或耗时)最短路线规划,采用MATLAB编程的方法来实现,并利用Floyd算法记录距离(或耗时)最短路线.在不考虑各种影响因素的情况下,以随机小样本数据为例进行演示,求得由起 ...
基于Twitter数据的情感预测与案例分析
导读本次分享的是关于文本分析中的情感预测分析和主题分析的一个小科研项目,数据爬取自Twitter,主要内容分为3部分: 1.情感预测机器学习模型选择及建立 2.情感预测深度学习模型LSTM的介绍和建 ...
手机信息采集 ——基于ebay网站Apple手机案例分析
手机信息采集 --基于ebay网站Apple手机在本案例分析中,我们要采集的是ebay网站上销售们的众多手机的价格.消费者评分和产品识别码等数据.ebay销售的产品很广,让我们能对每家大型手机生产商 ...
多线程-使用大全基础使用 / 锁 / 线程池 / 原子类 / 并发包 / CAS / AQS (2022版)
一.多线程描述 1.什么是cpu CPU的中文名称是中央处理器,是进行逻辑运算用的主要由运算器.控制器.寄存器三部分组成, 运算器:从字面意思看就是运算就是起着运算的作用, 控制器:就是负责发出cpu ...
Java并发包基石-AQS详解
目录 1 基本实现原理 1.1 如何使用 1.2 设计思想 2 自定义同步器 2.1 同步器代码实现 2.2 同步器代码测试 3 源码分析 3.1 Node结点 3.2 独占式 3.3 共享式 4 总 ...