- 浏览: 200824 次
- 性别:
- 来自: 上海
文章分类
最新评论
java之JUC系列-外部Tools-Executors|Semaphor|Exchanger|CyclicBarrier|CountDownLatch
前面写了两篇JDBC源码的文章,自己都觉得有点枯燥,先插一段JUC系列的文章来换换胃口,前面有文章大概介绍过J U C包含的东西,JUC体系包含的内容也是非常的多,不是一两句可以说清楚的,我这首先列出将会列举的JUC相关的内容,然后介绍本文的版本:Tools部分
J.U.C体系的主要大板块包含内容,如下图所示:
注意这个里面每个部分都包含很多的类和处理器,而且是相互包含,相互引用的,相互实现的。
说到J UC其实就是说java的多线程等和锁,前面说过一些状态转换,中断等,我们今天来用它的tools来实现一些有些小意思的东西,讲到其他内容的时候,再来想想这写tools是怎么实现的。
tools是本文说要讲到的重点,而tools主要包含哪些东西呢:
Tools也包含了5个部分的知识:Executors、Semaphor、Exchanger、CyclicBarrier、CountDownLatch,其实也就是五个工具类,这5个工具类有神马用途呢,就是我们接下来要将的内容了。
Executors:
其实它主要用来创建线程池,代理了线程池的创建,使得你的创建入口参数变得简单,通过方法名便知道了你要创建的线程池是什么样一个线程池,功能大概是什么样的,其实线程池内部都是统一的方法来实现,通过构造方法重载,使得实现不同的功能,但是往往这种方式很多时候不知道具体入口参数的改变有什么意思,除非读了源码才知道,此时builder模式的方式来完成,builder什么样的东西它告诉你就可以。
常见的方法有(都是静态方法):
1、创建一个指定大小的线程池,如果超过大小,放入blocken队列中,默认是LinkedBlockingQueue,默认的ThreadFactory为:Executors.defaultThreadFactory(),是一个Executors的一个内部类。
Executors.newFixedThreadPool(int)
内部实现是:
public static ExecutorService newFixedThreadPool(int nThreads) { return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>()); }
Executors.newFixedThreadPool(int,ThreadFactory)
内部实现是:
public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) { return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(), threadFactory); }
3、创建线程池长度为1的,也就是只有一个长度的线程池,多余的必须等待,它和调用Executors.newFixedThreadPool(1)得到的结果一样:
Executors.newSingleThreadExecutor()
内部实现是:
public static ExecutorService newSingleThreadExecutor() { return new FinalizableDelegatedExecutorService (new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>())); }是不是蛮简单的,就是在变参数,你自己也可以new的。
4、和方法3类似,可以自定义ThreadFactory,这里就不多说了!
5、创建可以进行缓存的线程池,默认缓存60s,数据会放在一个SynchronousQueue上,而不会进入blocken队列中,也就是只要有线程进来就直接进入调度,这个不推荐使用,因为容易出问题,除非用来模拟一些并发的测试:
Executors.newCachedThreadPool();
内部实现为:
public static ExecutorService newCachedThreadPool() { return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>()); }6、和方法5类似,增加自定义ThreadFactory
7、添加一个Schedule的调度器的线程池,默认只有一个调度:
Executors.newSingleThreadScheduledExecutor();
内部实现为(这里可以看到不是用ThreadPoolExector了,schedule换了一个类,内部实现通过ScheduledThreadPoolExecutor类里面的内部类ScheduledFutureTask来实现的,这个内部类是private,默认是引用不到的哦):
public static ScheduledExecutorService newSingleThreadScheduledExecutor() { return new DelegatedScheduledExecutorService (new ScheduledThreadPoolExecutor(1)); }8、和7一样,增加自己定义的ThreadFactory
9、添加一个schedule的线程池调度器,和newFixedThreadPool有点类似:
Executors.newScheduledThreadPool();
内部代码为:
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) { return new ScheduledThreadPoolExecutor(corePoolSize); }
其实内部Exectors里面还有一些其他的方法,我们就不多说明了,另外通过这里,大家先可以了解一个大概,知道Exectors其实是一个工具类,提供一系列的静态方法,来完成对对应线程池的形象化创建,所以不用觉得很神奇,神奇的是内部是如何实现的,本文我们不阐述文章中各种线程池的实现,只是大概上有个认识,等到我们专门将Exector系列的时候,我们会详细描述这些细节。
OK,我们继续下一个话题:
Semaphor,这个鸟东西是敢毛吃的呢?
答:通过名字就看出来了,是信号量。
信号量可以干什么呢?
答:根据一些阀值做访问控制。
OK,我们这里模拟一个当多个线程并发一段代码的时候,如何控制其访问速度:
import java.util.Random; import java.util.concurrent.Semaphore; public class SemaphoreTest { private final static Semaphore MAX_SEMA_PHORE = new Semaphore(10); public static void main(String []args) { for(int i = 0 ; i < 100 ; i++) { final int num = i; final Random radom = new Random(); new Thread() { public void run() { boolean acquired = false; try { MAX_SEMA_PHORE.acquire(); acquired = true; System.out.println("我是线程:" + num + " 我获得了使用权!" + DateTimeUtil.getDateTime()); long time = 1000 * Math.max(1, Math.abs(radom.nextInt() % 10)); Thread.sleep(time); System.out.println("我是线程:" + num + " 我执行完了!" + DateTimeUtil.getDateTime()); }catch(Exception e) { e.printStackTrace(); }finally { if(acquired) { MAX_SEMA_PHORE.release(); } } } }.start(); } } }
接下来:
Exchanger十个神马鬼东西呢?
答:线程之间交互数据,且在并发时候使用,两两交换,交换中不会因为线程多而混乱,发送出去没接收到会一直等,由交互器完成交互过程。
啥时候用,没想到案例?
答:的确很少用,而且案例很少,不过的确有这种案例,Exchanger
import java.util.concurrent.Exchanger; public class ExchangerTest { public static void main(String []args) { final Exchanger <Integer>exchanger = new Exchanger<Integer>(); for(int i = 0 ; i < 10 ; i++) { final Integer num = i; new Thread() { public void run() { System.out.println("我是线程:Thread_" + this.getName() + "我的数据是:" + num); try { Integer exchangeNum = exchanger.exchange(num); Thread.sleep(1000); System.out.println("我是线程:Thread_" + this.getName() + "我原先的数据为:" + num + " , 交换后的数据为:" + exchangeNum); } catch (InterruptedException e) { e.printStackTrace(); } } }.start(); } } }
这里运行你可以看到,如果某个线程和另一个线程传送了数据,它接受到的数据必然是另一个线程传递给他的,中间步骤由Exchanger去控制,其实你可以说,我自己随机取选择,不过中间的算法逻辑就要复杂一些了。
接下来:
CyclicBarrier,关卡模式,搞啥玩意的呢?
答:当你在很多环节需要卡住,要多个线程同时在这里都达到后,再向下走,很有用途。
能否举个例子,有点抽象?
答:团队出去旅行,大家一起先达到酒店住宿,然后一起达到游乐的地方游玩,然后一起坐车回家,每次需要点名后确认相关人员均达到,然后LZ一声令下,触发,大伙就疯子般的出发了。
下面的例子也是以旅游的方式来呈现给大家:
import java.util.concurrent.BrokenBarrierException; import java.util.concurrent.CyclicBarrier; public class BarrierTest { private static final int THREAD_COUNT = 10; private final static CyclicBarrier CYCLIC_BARRIER = new CyclicBarrier(THREAD_COUNT , new Runnable() { public void run() { System.out.println("======>我是导游,本次点名结束,准备走下一个环节!"); } } ); public static void main(String []args) throws InterruptedException, BrokenBarrierException { for(int i = 0 ; i < 10 ; i++) { new Thread(String.valueOf(i)) { public void run() { try { System.out.println("我是线程:" + this.getName() + " 我们达到旅游地点!"); CYCLIC_BARRIER.await(); System.out.println("我是线程:" + this.getName() + " 我开始骑车!"); CYCLIC_BARRIER.await(); System.out.println("我是线程:" + this.getName() + " 我们开始爬山!"); CYCLIC_BARRIER.await(); System.out.println("我是线程:" + this.getName() + " 我们回宾馆休息!"); CYCLIC_BARRIER.await(); System.out.println("我是线程:" + this.getName() + " 我们开始乘车回家!"); CYCLIC_BARRIER.await(); System.out.println("我是线程:" + this.getName() + " 我们到家了!"); } catch (InterruptedException e) { e.printStackTrace(); } catch (BrokenBarrierException e) { e.printStackTrace(); } } }.start(); } } }
测试结果中可以发现,大家一起走到某个步骤后,导游说:“我是导游,本次点名结束,准备走下一个环节!”,然后才会进入下一个步骤,OK,这个有点意思吧,其实赛马也是这个道理,只是赛马通常只有一个步骤,所以我们还有一个方式是:
CountDownLatch的方式来完成赛马操作,CountDownLatch是用计数器来做的,所以它不可以被复用,如果要多次使用,就要从新new一个出来才可以。我们下面的代码中,用两组赛马,每组5个参与者来,做一个简单测试:
import java.util.concurrent.CountDownLatch; public class CountDownLatchTest { private final static int GROUP_SIZE = 5; public static void main(String []args) { processOneGroup("分组1"); processOneGroup("分组2"); } private static void processOneGroup(final String groupName) { final CountDownLatch start_count_down = new CountDownLatch(1); final CountDownLatch end_count_down = new CountDownLatch(GROUP_SIZE); System.out.println("==========================>\n分组:" + groupName + "比赛开始:"); for(int i = 0 ; i < GROUP_SIZE ; i++) { new Thread(String.valueOf(i)) { public void run() { System.out.println("我是线程组:【" + groupName + "】,第:" + this.getName() + " 号线程,我已经准备就绪!"); try { start_count_down.await();//等待开始指令发出即:start_count_down.countDown(); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("我是线程组:【" + groupName + "】,第:" + this.getName() + " 号线程,我已执行完成!"); end_count_down.countDown(); } }.start(); } try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("各就各位,预备!"); start_count_down.countDown();//开始赛跑 try { end_count_down.await();//等待多个赛跑者逐个结束 } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("分组:" + groupName + "比赛结束!"); } }
public class ThreadJoinTest { private final static int GROUP_SIZE = 5; public static void main(String []args) throws InterruptedException { Thread []threadGroup1 = new Thread[5]; Thread []threadGroup2 = new Thread[5]; for(int i = 0 ; i < GROUP_SIZE ; i++) { final int num = i; threadGroup1[i] = new Thread() { public void run() { int j = 0; while(j++ < 10) { System.out.println("我是1号组线程:" + num + " 这个是我第:" + j + " 次运行!"); } } }; threadGroup2[i] = new Thread() { public void run() { int j = 0; while(j++ < 10) { System.out.println("我是2号组线程:" + num + " 这个是我第:" + j + " 次运行!"); } } }; threadGroup1[i].start(); } for(int i = 0 ; i < GROUP_SIZE ; i++) { threadGroup1[i].join(); } System.out.println("-==================>线程组1执行完了,该轮到俺了!"); for(int i = 0 ; i < GROUP_SIZE ; i++) { threadGroup2[i].start(); } for(int i = 0 ; i < GROUP_SIZE ; i++) { threadGroup2[i].join(); } System.out.println("全部结束啦!哈哈,回家喝稀饭!"); } }
代码是不是繁杂了不少,呵呵,我们再看看上面的信号量,如果不用工具,自己写会咋写,我们模拟CAS锁,使用Atomic配合完成咋来做呢。也来玩玩,呵呵:
import java.util.concurrent.atomic.AtomicInteger; public class ThreadWaitNotify { private final static int THREAD_COUNT = 100; private final static int QUERY_MAX_LENGTH = 2; private final static AtomicInteger NOW_CALL_COUNT = new AtomicInteger(0); public static void main(String []args) throws InterruptedException { Thread []threads = new Thread[THREAD_COUNT]; for(int i = 0 ; i < THREAD_COUNT ; i++) { threads[i] = new Thread(String.valueOf(i)) { synchronized public void run() { int nowValue = NOW_CALL_COUNT.get(); while(true) { if(nowValue < QUERY_MAX_LENGTH && NOW_CALL_COUNT.compareAndSet(nowValue, nowValue + 1)) { break;//获取到了 } try { this.wait(1000); } catch (InterruptedException e) { e.printStackTrace(); } nowValue = NOW_CALL_COUNT.get();//获取一个数据,用于对比 } System.out.println(this.getName() + "======我开始做操作了!"); try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println(this.getName() + "======操作结束了!"); NOW_CALL_COUNT.getAndDecrement(); this.notify(); } }; } for(int i = 0 ; i < THREAD_COUNT ; i++) { threads[i].start(); } } }
还是有点意思哈,这样写就是大部分人对while循环那部分会写晕掉,主要是要不断去判定和尝试,wait()默认是长期等待,但是我们不想让他长期等待,就等1s然后再尝试,其实例子还可以改成wait一个随机的时间范围,这样模拟的效果会更加好一些;另外实际的代码中,如果获取到锁后,notify方法应当放在finally中,才能保证他肯定会执行notify这个方法。
OK,本文就是用,玩,希望玩得有点爽,我们后面会逐步介绍它的实现机制以及一写线程里头很好用,但是大家又不是经常用的东西。
相关推荐
java - juc - 多线程 - 学习 -思维导图
本篇内容将详细介绍JUC中的一些核心组件及其使用方法,包括CountDownLatch、Executors、Semaphore、Exchanger、CyclicBarrier、LinkedBlockingQueue、ThreadFactory等。 首先,我们来探讨一下Executors工具类。...
Java-JUC-多线程进阶 Java-JUC-多线程进阶resources是 Java 并发编程的高级课程,涵盖了 Java 中的并发编程概念、线程安全、锁机制、集合类、线程池、函数式接口、Stream流式计算等多个方面。 什么是JUC JUC...
《juc尚硅谷-自学笔记》是一份针对Java并发编程领域的深入学习资料,主要涵盖了Java Concurrency Utility(juc)包中的核心概念和技术。在这个压缩包中,"尚硅谷juc"很可能包含了视频教程、课件或者讲义,旨在帮助...
本项目"tuling-juc-final.zip"显然聚焦于Java并发编程的实践,通过一系列代码示例来演示和解释Java内存模型(JMM)、`synchronized`关键字以及`volatile`关键字的使用。下面我们将深入探讨这些核心概念。 Java内存...
Java并发包(java.util.concurrent,简称JUC)提供了一系列工具和类库来帮助开发者简化并发编程的工作。其中,AbstractQueuedSynchronizer(简称AQS)是构建各种同步器的核心组件。 AQS是一个抽象的队列同步器,它...
### Java多线程与并发(15-26)-JUC集合-ConcurrentLinkedQueue详解 #### 一、ConcurrentLinkedQueue概述 `ConcurrentLinkedQueue`是Java实用工具包(J.U.C)中的一个高性能线程安全队列,主要用于解决多线程环境下...
### Java多线程与并发(17-26)-JUC线程池-FutureTask详解 #### 一、概述 本文将围绕Java多线程与并发中的重要概念——`FutureTask`进行深入探讨。`FutureTask`是Java并发库中的一个关键组件,它实现了`...
### Java多线程与并发(14-26)-JUC集合-CopyOnWriteArrayList详解 #### 一、概述 `CopyOnWriteArrayList`是一种线程安全的集合类,它是`ArrayList`的一种特殊版本,主要通过复制底层数组的方式来保证线程安全性。...
JUC的同步器包括CountDownLatch、CyclicBarrier、Semaphore等,它们是为了解决多个线程协作完成特定任务的工具。例如,CountDownLatch可以让一些线程等待其他线程完成操作后再继续执行,而CyclicBarrier则让一组线程...
Java 并发编程专题(九)----(JUC)浅析 CyclicBarrier CyclicBarrier 是 Java 并发编程中的一个同步辅助工具,它允许一组线程全部等待彼此到达公共屏障点。它的字面意思是可循环使用的屏障,用于让一组线程到达一...
Java中的ReentrantLock是Java并发包(java.util.concurrent.locks)中的一个高级锁,它是可重入的,意味着一个线程可以多次获取同一锁。在深入ReentrantLock之前,我们首先需要了解Java并发编程的基础,特别是Java...
在Java多线程并发编程中,ReentrantReadWriteLock(可重入读写锁)是一个重要的同步工具,它属于Java并发包(java.util.concurrent.locks)中的一个类。这个锁提供了比标准的synchronized关键字更细粒度的控制,允许...
JUC 框架包含多个部分,包括 Lock 框架、Tools 类、Collections、Atomic、Executors 等。下面是每个部分的详细解释: 1.Lock 框架和 Tools 类: Lock 框架提供了一种更灵活的锁定机制,可以代替 synchronized 方法...
这个类位于java.util.concurrent.locks包下,是实现并发编程中AQS(AbstractQueuedSynchronizer)框架的重要基础之一。LockSupport的目的是为了简化锁和其他同步类的实现,允许开发者创建更加高级的并发工具。 ...
Java中的`BlockingQueue`接口是Java并发编程的重要组件,它位于`java.util.concurrent`包下,主要用于线程间数据的同步和通信。`BlockingQueue`是一个支持线程安全的队列,它允许一个线程放入(插入)元素,而另一个...
【Java 多线程与并发】并发集合类`ConcurrentHashMap`是Java程序设计中一个重要的工具,尤其在高并发场景下,它提供了高效的线程安全。`ConcurrentHashMap`在JDK 1.7和1.8中有着显著的区别。 在JDK 1.7中,`...
Java并发编程是Java开发中的重要领域,而Java并发工具包(Java Concurrency Utility,简称JUC)则是Java标准库提供的一套强大而丰富的工具,它极大地简化了多线程环境下的编程工作。JUC主要包含在`java.util....
- JUC包含了许多高级并发组件,如`Semaphore`信号量、`CyclicBarrier`回环栅栏、`CountDownLatch`倒计时器和`Exchanger`交换器等,它们可以帮助控制线程的执行顺序。 - `Semaphore`可以限制同时访问特定资源的线程...