高效并发(四)

concurrent包总结(二)

前言

       上篇已经分析了lock和concurrent提供的集合类包括阻塞队列和容器类。这里我们来介绍

Executor框架

线程池的作用

       线程池作用就是限制系统中执行线程的数量。根据系统的环境情况,可以自动或手动设置线程数量,达到运行的最佳效果;少了浪费了系统资源,多了造成系统拥挤效率不高。用线程池控制线程数量,其他线程 排队等候。一个任务执行完毕,再从队列的中取最前面的任务开始执行。若队列中没有等待进程,线程池的这一资源处于等待。当一个新任务需要运行时,如果线程 池中有等待的工作线程,就可以开始运行了;否则进入等待队列。

为什么要用线程池

  1. 减少了创建和销毁线程的次数,每个工作线程都可以被重复利用,可执行多个任务
  2. 可以根据系统的承受能力,调整线程池中工作线线程的数目,防止因为因为消耗过多的内存,而把服务器累趴下(每个线程需要大约1MB内存,线程开的越多,消耗的内存也就越大,最后死机)

Executor

       Java里面线程池的顶级接口是Executor,里面就有个execute方法,但是严格意义上讲Executor并不是一个线程池,而只是一个执行线程的工具。真正的线程池接口是ExecutorService。ThreadPoolExecutor是Executors类的底层实现。

ExecutorService

shutDown()

       当线程池调用该方法时,线程池的状态则立刻变成SHUTDOWN状态。此时,则不能再往线程池中添加任何任务,否则将会抛出RejectedExecutionException异常。但是,此时线程池不会立刻退出,直到添加到线程池中的任务都已经处理完成,才会退出。

shutdownNow()

       根据JDK文档描述,大致意思是:执行该方法,线程池的状态立刻变成STOP状态,并试图停止所有正在执行的线程,不再处理还在池队列中等待的任务,当然,它会返回那些未执行的任务。

       它试图终止线程的方法是通过调用Thread.interrupt()方法来实现的,但是大家知道,这种方法的作用有限,如果线程中没有sleep 、wait、Condition、定时锁等应用, interrupt()方法是无法中断当前的线程的。所以,ShutdownNow()并不代表线程池就一定立即就能退出,它可能必须要等待所有正在执行的任务都执行完成了才能退出。

其他方法

       包括shubmit,invoke相关和Terminate相关。我们在实现类中在继续分析。

ThreadPoolExecutor

final AtomicInteger ctl

       这个变量是整个类的核心,AtomicInteger保证了对这个变量的操作是原子的,通过巧妙的操作,ThreadPoolExecutor用这一个变量保存了两个内容:

  1. 所有有效线程的数量
  2. 各个线程的状态(runState)

低29位存线程数,高3位存runState,这样runState有5个值,解释如下:

//线程池正常运行,可以接受新的任务并处理队列中的任务;
private static final int RUNNING    = -1 << COUNT_BITS;
//不再接受新的任务,但是会执行队列中的任务;
private static final int SHUTDOWN   =  0 << COUNT_BITS;
//不再接受新任务,不处理队列中的任务
private static final int STOP       =  1 << COUNT_BITS;
private static final int TIDYING    =  2 << COUNT_BITS;
private static final int TERMINATED =  3 << COUNT_BITS;

// Packing and unpacking ctl
private static int runStateOf(int c)     { return c & ~CAPACITY; }
private static int workerCountOf(int c)  { return c & CAPACITY; }
private static int ctlOf(int rs, int wc) { return rs | wc; }

内部封装了work类

       内部类Worker是对任务的封装,所有submit的Runnable都被封装成了Worker,它本身也是一个Runnable, 然后利用AQS框架(关于AQS可以看我这篇文章)实现了一个简单的非重入的互斥锁, 实现互斥锁主要目的是为了中断的时候判断线程是在空闲还是运行。

Executors

       有4个重要实现类:

  1. newCachedThreadPool创建一个可缓存线程池,如果线程池长度超过处理需要,可灵活回收空闲线程,若无可回收,则新建线程。
  2. newFixedThreadPool 创建一个定长线程池,可控制线程最大并发数,超出的线程会在队列中等待。
  3. newScheduledThreadPool 创建一个定长线程池,支持定时及周期性任务执行。
  4. newSingleThreadExecutor 创建一个单线程化的线程池,它只会用唯一的工作线程来执行任务,保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。

同步工具类

       还有号称四大金刚的同步器。同步工具类可以使任何一种对象,只要该对象可以根据自身的状态来协调控制线程的控制流。阻塞队列可以作为同步工具类,其他类型的同步工具类还包括:信号量(Semaphore)、栅栏(Barrier)以及闭锁(Latch)和交换(exchanger)参考博文:同步工具类解析

CountDownLatch闭锁

首先我们来介绍闭锁。

       闭锁作用相当于一扇门:在闭锁到达某一状态之前,这扇门一直是关闭的,所有的线程都会在这扇门前等待(阻塞)。只有门打开后,所有的线程才会同时继续运行。
闭锁可以用来确保某些活动直到其它活动都完成后才继续执行,例如:

  1. 确保某个计算在其所有资源都被初始化之后才继续执行。二元闭锁(只有两个状态)可以用来表示“资源R已经被初始化”,而所有需要R操作都必须先在这个闭锁上等待。
  2. 确保某个服务在所有其他服务都已经启动之后才启动。这时就需要多个闭锁。让S在每个闭锁上等待,只有所有的闭锁都打开后才会继续运行。
    3。 等待直到某个操作的参与者(例如,多玩家游戏中的玩家)都就绪再继续执行。在这种情况下,当所有玩家都准备就绪时,闭锁将到达结束状态。

       CountDownLatch 是一种灵活的闭锁实现,可以用在上述各种情况中使用。闭锁状态包含一个计数器,初始化为一个正数,表示要等待的事件数量。countDown() 方法会递减计数器,表示等待的事件中发生了一件。await() 方法则阻塞,直到计数器值变为0。

       下面,我们使用闭锁来实现在主线程中计算多个子线程运行时间的功能。具体逻辑是使用两个闭锁,“起始门”用来控制子线程同时运行,“结束门”用来标识子线程是否都结束。

CyclicBarrier和exchanger

       栅栏(Bariier)类似于闭锁,它能阻塞一组线程知道某个事件发生。栅栏与闭锁的关键区别在于,所有的线程必须同时到达栅栏位置,才能继续执行。闭锁用于等待等待时间,而栅栏用于等待线程。

       CyclicBarrier 可以使一定数量的参与方反复的在栅栏位置汇聚,它在并行迭代算法中非常有用:将一个问题拆成一系列相互独立的子问题。当线程到达栅栏位置时,调用await() 方法,这个方法是阻塞方法,直到所有线程到达了栅栏位置,那么栅栏被打开,此时所有线程被释放,而栅栏将被重置以便下次使用。

       另一种形式的栅栏是Exchanger,它是一种两方(Two-Party)栅栏,各方在栅栏位置上交换数据。例如当一个线程想缓冲区写入数据,而另一个线程从缓冲区中读取数据。这些线程可以使用 Exchanger 来汇合,并将慢的缓冲区与空的缓冲区交换。当两个线程通过 Exchanger 交换对象时,这种交换就把这两个对象安全的发布给另一方。

       Exchanger 可能被视为 SynchronousQueue 的双向形式。我们也可以用两个SynchronousQueue来实现 Exchanger的功能。

信号量Semaphone

       之前讲的闭锁控制访问的时间,而信号量则用来控制访问某个特定资源的操作数量,控制空间。而且闭锁只能够减少,一次性使用,而信号量则申请可释放,可增可减。 计数信号量还可以用来实现某种资源池,或者对容器施加边界。

       Semaphone 管理这一组许可(permit),可通过构造函数指定。同时提供了阻塞方法acquire,用来获取许可。同时提供了release方法表示释放一个许可。

       Semaphone 可以将任何一种容器变为有界阻塞容器,如用于实现资源池。例如数据库连接池。我们可以构造一个固定长度的连接池,使用阻塞方法 acquire和release获取释放连接,而不是获取不到便失败。(当然,一开始设计时就使用BlockingQueue来保存连接池的资源是一种更简单的方法)。

FutureTask

       FutureTask也可以用作闭锁。它表示一种抽象的可生成结果的计算。是通过 Callable 来实现的,相当于一种可生成结果的 Runnable ,并且可处于以下三种状态:等待运行,正在运行,运行完成。当FutureTask进入完成状态后,它会停留在这个状态上。

       Future.get 用来获取计算结果,如果FutureTask还未运行完成,则会阻塞。FutureTask 将计算结果从执行计算的线程传递到获取这个结果的线程,而FutureTask 的规范确保了这种传递过程能实现结果的安全发布。

       FutureTask在Executor框架中表示异步任务,还可以用来表示一些时间较长的计算,这些计算可以在使用计算结果之前启动。

说明

       文中出现的图片,文字描述有些来自互联网,但是出处无法考究,如果侵犯您的相关权益,请联系我,核实后我会马上加上转载说明。谢谢!!!

坚持原创技术分享,您的支持将鼓励我继续创作!