ConcurrentInJava

这一系列只是对JUC各个部分做了说明和介绍，没人深入原理!

创新互联公司主要从事成都网站制作、做网站、网页设计、企业做网站、公司建网站等业务。立足成都服务嵩县,10年网站建设经验,价格优惠、服务专业,欢迎来电咨询建站服务:13518219792

concurrent并发包，让你易于编写并发程序。并发下我们经常需要使用的基础设施和解决的问题有ThreadPool、Lock、管道、集合点、线程之间等待和唤醒、线程间数据传输、共享资源访问控制、并发线程之间的相互等待，等待。

concurrent提供的工具能够解决绝大部分的场景，还能提高程序吞吐量。

现代的服务器多采用多核CPU，从而不同线程之间有可能真正地在同时运行而不是cpu时间切片。在处理大计算量的程序上要尽可能利用CPU多核特性，提高系统吞吐量。

并发编程主要面临三个问题：

1.如何让多个线程同时为同一个任务工作(并发编程设计)

2.多个线程之间对共享资源的争用。

3.多个线程之间如何相互合作、传递数据。

1. concurrent包提供的集合

concurrent包直接提供了标准集合的一些实现，在下面做简单介绍。在大部分情况下可以使用它们提供高并发环境下对集合访问的吞吐量。

1.1 ConcurrentHashMap

Map的一个并发实现。在多线程环境下，它具有很高的吞吐量和具备可靠的数据一致性。它支持并发读和一定程度的并发修改(默认16个并发，可以通过构造函数修改)。

HashMap的实现是非线程安全的，高并发下会get方法常会死锁，有的时候会表现为CPU居高不下。

 
 
 
 
    
  
  
  
  
  1. public V get(Object key) {  
    2. 
  
  
  
  
  2. if (key == null)  
    3. 
  
  
  
  
  3. return getForNullKey();  
    4. 
  
  
  
  
  4. int hash = hash(key.hashCode());  
    5. 
  
  
  
  
  5. for (Entry e = table[indexFor(hash, table.length)];  
    6. 
  
  
  
  
  6. e != null;  
    7. 
  
  
  
  
  7. e = e.next) {  
    8. 
  
  
  
  
  8. Object k;  
    9. 
  
  
  
  
  9. if (e.hash == hash && ((k = e.key) == key || key.equals(k)))  
    10. 
  
  
  
  
  10. return e.value;  
    11. 
  
  
  
  
  11. }  
    12. 
  
  
  
  
  12. return null;  
    13. 
  
  
  
  
  13. }  
    14. 
  
  
  
  
  14.  
    15.

在get操作里面for循环取对象的操作，由于高并发同时读写，for循环的结果变得不可预知，所以有可能一直循环。

所以高并发环境下尽量不要直接使用HashMap，对系统造成的影响很难排除。

和Collections.synchronizedMap(new HashMap(...))相比，外ConcurrentHashMap在高并发的环境下有着更优秀的吞吐量。因为ConcurrentHashMap可以支持写并发，基本原理是内部分段，分段的数量决定着并发程度。通过concurrencyLevel参数可以设置。如果你能预期并发数量那么设置该参数可以获取更优吞吐量。

另外为ConcurrentHashMap还实现了：

V putIfAbsent(K key, V value);

boolean remove(Object key, Object value);

boolean replace(K key, V oldValue, V newValue);

V replace(K key, V value);

这四个一致性的操作方法。

1.2 BlockingQueue

BlockingQueue定义了一个接口，继承了Queue接口。Queue是一种数据结构，意思是它的项以先入先出(FIFO)顺序存储。

BlockingQueue为我们提供了一些多线程阻塞语义的方法，新增和重定义了一些方法插入：

BlockingQueue是线程安全的，非常适合多个生产者和多个消费者线程之间传递数据。

形象地理解，BlockingQueue好比有很多格子的传输带系统，不过当你(生产者)调用put方法的时候，如果有空闲的格子那么放入物体后立刻返回，如果没有空闲格子那么一直处于等待状态。add方法意味着如果没有空闲格子系统就会报警，然后如果处理该报警则按照你的意愿。offer方法优先于add方法，它通过返回true 或 flase来告诉你是否放入成功。offer超时方法，如果不空闲的情况下，尝试等待一段时间。

BlockingQueue有很多实现ArrayBlockingQueue, DelayQueue, LinkedBlockingDeque, LinkedBlockingQueue, PriorityBlockingQueue, SynchronousQueue

补充Dueue是个双向队列，可以当做堆栈来使用。

BlockingQueue在ThreadPool中，作为任务队列来使用，用来保存没有立刻执行的工作任务对象。

1.3 SynchronousQueue

SychronousQueue是BlockingQueue的一个实现，它看起来是一个队列，但是其实没有容量，是特定条件下的一个精简实现。

做个比喻，SychronousQueue对象就像一个接力棒，现在有两个运动员交棒者和接棒者(线程)要做交接。在交接点，交棒者没有交出之前是不能松开的(一种等待状态)，接棒者在接到棒之前是必须等待。换一句话说不管谁先到交接点，必须处于等待状态。

在生产者和消费者模型中。如果生产者向SychronousQueue进行put操作，直到有另外的消费者线程进行take操作时才能返回。对消费者也是一样，take操作会被阻塞，直到生产者put。

在这种生产者-消费者模型下，生产者和消费者是进行手对手传递产品，在消费者消费一个产品之前，生产者必须处于等待状态。它给我们提供了在线程之间交换单一元素的极轻量级方法，并且具有阻塞语义。

提示：上面举例中有写局限性。其实生产者和消费者进程是可以任意数量的。M：N。生产线程之间会对SychronousQueue进行争用，消费者也是一样。

对SychronousQueue类似于其他语境中“会合通道”或 “连接”点问题。它非常适合于传递性设计，在这种设计中，在一个线程中运行的对象要将某些信息、事件或任务传递给在另一个线程中运行的对象，它就必须与该对象同步。

1.4Exchanger

是SychronousQueue的双向实现。用来伙伴线程间交互对象。Exchanger 可能在比如遗传算法和管道设计中很有用。

形象地说，就是两个人在预定的地方交互物品，任何一方没到之前都处于等待状态。

1.5 CopyOnWriteArrayList 和 CopyOnWriteArraySet

它们分别是List接口和Set接口的实现。正如类名所描述的那样，当数据结构发生变化的时候，会复制自身的内容，来保证一致性。大家都知道复制全部副本是非常昂贵的操作，看来这是一个非常不好的实现。事实上没有最好和最差的方案，只有最合适的方案。一般情况下，处理多线程同步问题，我们倾向使用同步的 ArrayList，但同步也有其成本。

那么在什么情况下使用CopyOnWriteArrayList 或者CopyOnWriteArraySet呢?

数据量小。

对数据结构的修改是偶然发生的，相对于读操作。

举例来说，如果我们实现观察者模式的话，作为监听器集合是非常合适的。

1.6 TimeUnit

虽然是个时间单位，但是它也是concurrent包里面的。也许你以前的代码里面经常出现1*60*1000来表示一分钟，代码可读性很差。现在你可以通过TimeUnit来编写可读性更好的代码，concurrent的api里面涉及到时间的地方都会使用该对象。

我之所以先进并发框架常用的集合，是因为线程池的实现特性都利用了BlockingQueue的一些特性。

#p#

2. ThreadPool

虽然线程和进程相比是轻量级许多，但是线程的创建成本还是不可忽律，所以就有了线程池化的设计。线程池的创建、管理、回收、任务队列管理、任务分配等细节问题依然负责，没有必要重复发明轮子，concurrent包已经为我们准备了一些优秀线程池的实现。

2.1 认识ExecutorService 接口

ExecutorService 接口，它能提供的功能就是用来在将来某一个时刻异步地执行一系列任务。虽然简单一句话，但是包含了很多需求点。它的实现至少包含了线程池和任务队列两个方面，其实还包括了任务失败处理策略等。

经常使用submit方法，用来提交任务对象。

简单的例子：

ExecutorService es = Executors.newCachedThreadPool();

es.submit(new Runnable(){

@Override

public void run() {

System.out.println("do some thing");

}

});

es.shutdown();

上面的例子只是完成了提交了一个任务，异步地去执行它。但是有些使用场景更为复杂，比如等待获得异步任务的返回结果，或者最多等上固定的时间。

submit 方法返回一个对象，Future。看起来有点别扭，代表将来的对象。其实看一下Future的方法就明白了。

其实Future对象代表了一个异步任务的结果，可以用来取消任务、查询任务状态，还有通过get方法获得异步任务返回的结果。当调用get方法的时候，当前线程被阻塞直到任务被处理完成或者出现异常。

我们可以通过保存Future对象来跟踪查询异步任务的执行情况。

显然Runnable接口中定义的 public void run();方法并不能返回结果对象，所以concurrent包提供了Callable接口，它可以被用来返回结果对象。

2.2 ThreadPoolExecutor

ThreadPoolExecutor实现了ExecutorService 接口，也是我们最主要使用的实现类。

首先非常有必要看一些类的最完整的构造函数

 
 
 
 
    
  
  
  
  
  1. ThreadPoolExecutor(int corePoolSize,  
    2. 
  
  
  
  
  2.  
    3. 
  
  
  
  
  3.    int maximumPoolSize,  
    4. 
  
  
  
  
  4.  
    5. 
  
  
  
  
  5. long keepAliveTime,  
    6. 
  
  
  
  
  6.  
    7. 
  
  
  
  
  7. TimeUnit unit,  
    8. 
  
  
  
  
  8.  
    9. 
  
  
  
  
  9. BlockingQueue  workQueue,   
    10. 
  
  
  
  
  10.  
    11. 
  
  
  
  
  11. ThreadFactory threadFactory,  
    12. 
  
  
  
  
  12.  
    13. 
  
  
  
  
  13. RejectedExecutionHandler handler)  
    14. 
  
  
  
  
  14.  
    15.

ThreadPoolExecutor对象中有个poolSize变量表示当前线程池中正在运行的线程数量。

注意：这个有关非常重要的关系，常常被误解。poolSize变量和corePoolSize、maximumPoolSize以及workQueue的关系。

首先线程池被创建初期，还没有执行任何任务的时候，poolSize等于0;

每次向线程池提交任务的时候，线程池处理过程如下：

1. 如果poolSize少于 corePoolSize，则首选添加新的线程，而不进行排队。

2. 如果poolSize等于或多于 corePoolSize，则首选将请求加入队列workQueue，而不添加新的线程。

3. 如果第二步执行失败(队已满)，则创建新的线程执行任务，但是如果果poolSize已经达到maximumPoolSize，那么就拒绝该任务。如果处理被拒绝的任务就取决于RejectedExecutionHandler handler的设置了，默认情况下会抛出异常。

系统存在四种任务拒绝策略：

在默认的 ThreadPoolExecutor.AbortPolicy 中，处理程序遭到拒绝将抛出运行时 RejectedExecutionException。

在 ThreadPoolExecutor.CallerRunsPolicy 中，线程调用运行该任务的 execute 本身。此策略提供简单的反馈控制机制，能够减缓新任务的提交速度。

在 ThreadPoolExecutor.DiscardPolicy 中，不能执行的任务将被删除。

在 ThreadPoolExecutor.DiscardOldestPolicy 中，如果执行程序尚未关闭，则位于工作队列头部的任务将被删除，然后重试执行程序(如果再次失败，则重复此过程)。

keepAliveTime活动线程如果空闲一段时间是否可以回收，通常只作用于超出corePoolSize的线程。corePoolSize的线程创建了就不会被回收。但是到java 6 之后增加了public void allowCoreThreadTimeOut(boolean value)方法，允许core进程也可以根据keepAliveTime来回收，默认为false。

决定线程池特性的还有workQueue的实现类，有三种类SynchronousQueue、LinkedBlockingQueue、ArrayBlockingQueue，分别对应同步队列、无界队列、有界队列。

(摘自JavaDoc)

类SynchronousQueue，直接提交。工作队列的默认选项是 SynchronousQueue，它将任务直接提交给线程而不保持它们。在此，如果不存在可用于立即运行任务的线程，则试图把任务加入队列将失败，因此会构造一个新的线程。此策略可以避免在处理可能具有内部依赖性的请求集时出现锁。直接提交通常要求无界 maximumPoolSizes 以避免拒绝新提交的任务(设置maximumPoolSizes 为Integer.MAX_VALUE)。当命令以超过队列所能处理的平均数连续到达时，此策略允许无界线程具有增长的可能性。

LinkedBlockingQueue，无界队列。使用无界队列(例如，不具有预定义容量的 LinkedBlockingQueue)将导致在所有 corePoolSize 线程都忙时新任务在队列中等待。这样，创建的线程就不会超过 corePoolSize。(因此，maximumPoolSize 的值也就无效了。)当每个任务完全独立于其他任务，即任务执行互不影响时，适合于使用无界队列;例如，在 Web 页服务器中。这种排队可用于处理瞬态突发请求，当命令以超过队列所能处理的平均数连续到达时，此策略允许无界线程具有增长的可能性。

ArrayBlockingQueue，有界队列。当使用有限的 maximumPoolSizes 时，有界队列(如 ArrayBlockingQueue)有助于防止资源耗尽，但是可能较难调整和控制。队列大小和最大池大小可能需要相互折衷：使用大型队列和小型池可以最大限度地降低 CPU 使用率、操作系统资源和上下文切换开销，但是可能导致人工降低吞吐量。如果任务频繁阻塞(例如，如果它们是 I/O 边界)，则系统可能为超过您许可的更多线程安排时间。使用小型队列通常要求较大的池大小，CPU 使用率较高，但是可能遇到不可接受的调度开销，这样也会降低吞吐量。

综上：构造参数的设置是互相制约和影响的。只有当你重复了解其相互关系的时候、或有特殊需求的时候，才可以自己构造ThreadPoolExecutor对象，否则可以使用Executores是个工厂类。

提示使用线程池是注意处理shutdown，确保你系统关闭的时候主动关闭shutdown。

2.3 ScheduledExecutorService

扩展了ExecutorService接口，提供时间排程的功能。

schedule方法被用来延迟指定时间来执行某个指定任务。如果你需要周期性重复执行定时任务可以使用scheduleAtFixedRate或者scheduleWithFixedDelay方法，它们不同的是前者以固定频率执行，后者以相对固定频率执行。

(感谢wenbois2000 提出原先的错误，我在这里重新描述!对于原先的错误，实在不好意思啊，再次感谢!)不管任务执行耗时是否大于间隔时间，scheduleAtFixedRate和scheduleWithFixedDelay都不会导致同一个任务并发地被执行。唯一不同的是scheduleWithFixedDelay是当前一个任务结束的时刻，开始结算间隔时间，如0秒开始执行第一次任务，任务耗时5秒，任务间隔时间3秒，那么第二次任务执行的时间是在第8秒开始。

ScheduledExecutorService的实现类，是ScheduledThreadPoolExecutor。ScheduledThreadPoolExecutor对象包含的线程数量是没有可伸缩性的，只会有固定数量的线程。不过你可以通过其构造函数来设定线程的优先级，来降低定时任务线程的系统占用。

特别提示：通过ScheduledExecutorService执行的周期任务，如果任务执行过程中抛出了异常，那么过ScheduledExecutorService就会停止执行任务，且也不会再周期地执行该任务了。所以你如果想保住任务都一直被周期执行，那么catch一切可能的异常。

2.4 Executors

Executores是个工厂类，用来生成ThreadPoolExecutor对象，它提供了一些常用的线程池配置方案，满足我们大部分场景。

1. newCachedThreadPool

 
 
 
 
    
  
  
  
  
  1. public static ExecutorService newCachedThreadPool() {  
    2. 
  
  
  
  
  2. return new ThreadPoolExecutor(0, Integer.MAX_VALUE,  
    3. 
  
  
  
  
  3. 60L, TimeUnit.SECONDS,  
    4. 
  
  
  
  
  4. new SynchronousQueue());  
    5. 
  
  
  
  
  5. }  
    6. 
  
  
  
  
  6.  
    7.

分析下出这个线程池配置的工作模式，当没有空闲进程时就新建线程执行，当有空闲线程时就使用空闲线程执行。当线程空闲大60秒时，系统自动回收线程。

该线程池非常适合执行短小异步任务时吞吐量非常高，会重复利用CPU的能力。但是如果任务处理IO边界任务，那么会消耗大量线程切换，降低系统吞吐量。所以执行短小的计算任务非常高效，且当没有任务时不会消耗系统资源。

注意：线程池中没有变量表示线程是否空闲。那么程序是如何控制的呢?不得不赞叹concurrent实现的非常精巧。当创建出来的线程完成原来的任务后，会调用BlockingQueue的Poll方法，对于SynchronousQueue实现而言会阻塞调用线程，直到另外的线程offer调用。

然而ThreadPool在分配任务的时候总是先去尝试调用offer方法，所以就会触发空闲线程再次调用。

精妙的是ThreadPoolExecutor的处理逻辑一样，但是用BlockingQueue实现变了就产生不同的行为。

2. newFixedThreadPool

 
 
 
 
    
  
  
  
  
  1. public static ExecutorService newFixedThreadPool(int nThreads) {  
    2. 
  
  
  
  
  2.  
    3. 
  
  
  
  
  3. return new ThreadPoolExecutor(nThreads, nThreads,  
    4. 
  
  
  
  
  4.  
    5. 
  
  
  
  
  5. 0L, TimeUnit.MILLISECONDS,  
    6. 
  
  
  
  
  6.  
    7. 
  
  
  
  
  7. new LinkedBlockingQueue());  
    8. 
  
  
  
  
  8.  
    9. 
  
  
  
  
  9. }  
    10. 
  
  
  
  
  10.  
    11.

创建固定线程数量的线程池，采用无界队列，当有更多任务的时候将被放入工作队列中排队。如果线程池不经常执行任务时，你可以调用allowCoreThreadTimeOut(boolean value)的方法让系统自动回收core的进程，以节约系统资源。

3. newSingleThreadExecutor

 
 
 
 
    
  
  
  
  
  1. public static ExecutorService newSingleThreadExecutor() {  
    2. 
  
  
  
  
  2.  
    3. 
  
  
  
  
  3. return new FinalizableDelegatedExecutorService  
    4. 
  
  
  
  
  4.  
    5. 
  
  
  
  
  5. (new ThreadPoolExecutor(1, 1,  
    6. 
  
  
  
  
  6.  
    7. 
  
  
  
  
  7. 0L, TimeUnit.MILLISECONDS,  
    8. 
  
  
  
  
  8.  
    9. 
  
  
  
  
  9. new LinkedBlockingQueue()));  
    10. 
  
  
  
  
  10.  
    11. 
  
  
  
  
  11. }  
    12. 
  
  
  
  
  12.  
    13.

只有一个工作线程的线程池。和newFixedThreadPool(1)相比，不同之处有两点：

1. 不可以重新配置newSingleThreadExecutor创建出来的线程池。

2. 当创建出来的线程池对象被GC回收时，会自动调用shutdown方法。

4.newScheduledThreadPool

 
 
 
 
    
  
  
  
  
  1. public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {  
    2. 
  
  
  
  
  2.        return new ScheduledThreadPoolExecutor(corePoolSize);  
    3. 
  
  
  
  
  3.    }  
    4.

生成一个可以执行时间调度的线程池。其实内部使用无界工作队列，线程数量最多能达到corePoolSize。

2.5 ExecutorCompletionService

这是个巧妙的设计，内部维护了一已经完成了任务结果队列，通过take方法可以同步地等待一个个结果对象。

详情见http://www.oschina.net/uploads/doc/javase-6-doc-api-zh_CN/java/util/concurrent/ExecutorCompletionService.html

#p#

3. java.util.concurrent.locks

java 早期内置synchronized关键字解决多线程对共享资源访问的一些问题,和其还配套了Object的notify 和 wait方法,用来控制线程之间的同步。

concurrent软件包提供了更为高级和抽象的Lock工具，能解决更多的问题。

Lock是控制多个线程对共享资源进行访问的工具。通常Lock限定多线程对同一个共享资源访问的限制，一次只允许一个线程获得Lock，即获得对共享资源的访问权限，线程间是互斥的。但是也有一些锁如果ReadWriteLock是允许部分线程同时访问共享资源的。

几个术语：

争用：当多个Thread在同一时间内(相对概念)想要占有同一个Lock对象。那么JVM会调度解决争用。

获取顺序：当多个线程争用同一个Lock对象，那么JVM就要决定哪个线程将会获得锁权限。存在两种模式：公平和不公平。 默认都是不公平模式，包括synchronized关键字，jvm决定顺序的时候也是采用不公平策略。因为公平策略需要系统记录大量辅助信息来判断分配顺序，而不公平策略由JVM决定一直快速高效的算法来分配Lock。所以不公平策略的系统吞吐量会比较高(花费更少的空间和计算在分配上)，如果没有特殊需要则默认采用不公平策略。

重入：当前线程获取指定的锁对象权限后，还可以再次获取该锁。Lock内部会有一个计数器来表明当前线程获取了该锁的数量。如果一个线程获取了一个锁两次，那么线程必须释放锁两次，才能被看作完全释放了该锁，所以编程的时候一定要注意使用重入。synchronized关键字也是支持重入语义的。

3.1 Lock & ReentrantLock

ReentrantLock实现了Lock接口,一个可重入(reentrant)的互斥锁 Lock，它具有与使用 synchronized 方法和语句所访问的隐式监视器锁相同的一些基本行为和语义，但功能更强大。

摘自JavaDoc的一段获取规则 “当锁没有被另一个线程所拥有时，调用 lock 的线程将成功获取该锁并返回。如果当前线程已经拥有该锁，此方法将立即返回。ReentrantLock 将由最近成功获得锁，并且还没有释放该锁的线程所拥有。”

经典使用方法。

 
 
 
 
    
  
  
  
  
  1. public void m() {   
    2. 
  
  
  
  
  2.     lock.lock();  // block until condition holds  
    3. 
  
  
  
  
  3.     try {  
    4. 
  
  
  
  
  4.       // ... method body  
    5. 
  
  
  
  
  5.     } finally {  
    6. 
  
  
  
  
  6.       lock.unlock()  
    7. 
  
  
  
  
  7.     }  
    8. 
  
  
  
  
  8.   }  
    9. 
  
  
  
  
  9.  
    10.

ReentrantLock除了实现了Lock规定的方法外，还实现了tryLock、isLocked 等方法，帮助你实现更多的场景。

Condition

和Object的wait和notify方法类似。ReentrantLock对象附加了Conditon对象，用来完成挂起和唤醒操作，使用lock.newCondition() 方法生成。

一个来自JKD的例子：

 
 
 
 
    
  
  
  
  
  1. class BoundedBuffer {  
    2. 
  
  
  
  
  2.   final Lock lock = new ReentrantLock();  
    3. 
  
  
  
  
  3.   final Condition notFull  = lock.newCondition();   
    4. 
  
  
  
  
  4.   final Condition notEmpty = lock.newCondition();   
    5. 
  
  
  
  
  5.  
    6. 
  
  
  
  
  6.   final Object[] items = new Object[100];  
    7. 
  
  
  
  
  7.   int putptr, takeptr, count;  
    8. 
  
  
  
  
  8.  
    9. 
  
  
  
  
  9.   public void put(Object x) throws InterruptedException {  
    10. 
  
  
  
  
  10.     lock.lock();  
    11. 
  
  
  
  
  11.     try {  
    12. 
  
  
  
  
  12.       while (count == items.length)   
    13. 
  
  
  
  
  13.         notFull.await();  
    14. 
  
  
  
  
  14.       items[putptr] = x;   
    15. 
  
  
  
  
  15.       if (++putptr == items.length) putptr = 0;  
    16. 
  
  
  
  
  16.       ++count;  
    17. 
  
  
  
  
  17.       notEmpty.signal();  
    18. 
  
  
  
  
  18.     } finally {  
    19. 
  
  
  
  
  19.       lock.unlock();  
    20. 
  
  
  
  
  20.     }  
    21. 
  
  
  
  
  21.   }  
    22. 
  
  
  
  
  22.  
    23. 
  
  
  
  
  23.   public Object take() throws InterruptedException {  
    24. 
  
  
  
  
  24.     lock.lock();  
    25. 
  
  
  
  
  25.     try {  
    26. 
  
  
  
  
  26.       while (count == 0)   
    27. 
  
  
  
  
  27.         notEmpty.await();  
    28. 
  
  
  
  
  28.       Object x = items[takeptr];   
    29. 
  
  
  
  
  29.       if (++takeptr == items.length) takeptr = 0;  
    30. 
  
  
  
  
  30.       --count;  
    31. 
  
  
  
  
  31.       notFull.signal();  
    32. 
  
  
  
  
  32.       return x;  
    33. 
  
  
  
  
  33.     } finally {  
    34. 
  
  
  
  
  34.       lock.unlock();  
    35. 
  
  
  
  
  35.     }  
    36. 
  
  
  
  
  36.   }   
    37. 
  
  
  
  
  37. }  
    38.

利用Conditon对象可以让所有对同一个锁对象进行争用的Thread之间进行同步。

Lock VS synchronized

除非你有明确的需求或者并发遇到瓶颈的时候再决定使用ReentrantLock。synchronized在大部分时候还是可以工作的很好，jvm会自动处理和回收锁。

ReentrantLock提供了更多的选择和状态信息。和

3.2 ReadWriteLock & ReentrantReadWriteLock

列举一个场景对象X，拥有方法a、b、c。a和b方法不改表X的内部状态，c改变内部状态。在多线程环境下，我们要求只读和写(变更状态)是不能同时进行的，而只读操作是可以同时并发的,且实际运行过程中读操作数量远远大于写操作的数量。

如果用synchronized关键字的话，两个只读方法a、b也会互斥，并发性能收到限制。

那么这个情况下ReadWriteLock就非常有用，使用也非常简单。

 
 
 
 
    
  
  
  
  
  1. class RWDictionary {  
    2. 
  
  
  
  
  2.    private final Map  m =  new TreeMap ();    
    3. 
  
  
  
  
  3.    private final ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();  
    4. 
  
  
  
  
  4.    private final Lock r = rwl.readLock();  
    5. 
  
  
  
  
  5.    private final Lock w = rwl.writeLock();  
    6. 
  
  
  
  
  6.  
    7. 
  
  
  
  
  7.    public Data get(String key) {  
    8. 
  
  
  
  
  8.        r.lock();  
    9. 
  
  
  
  
  9.        try { return m.get(key); }  
    10. 
  
  
  
  
  10.        finally { r.unlock(); }  
    11. 
  
  
  
  
  11.    }  
    12. 
  
  
  
  
  12.    public String[] allKeys() {  
    13. 
  
  
  
  
  13.        r.lock();  
    14. 
  
  
  
  
  14.        try { return m.keySet().toArray(); }  
    15. 
  
  
  
  
  15.        finally { r.unlock(); }  
    16. 
  
  
  
  
  16.    }  
    17. 
  
  
  
  
  17.    public Data put(String key, Data value) {  
    18. 
  
  
  
  
  18.        w.lock();  
    19. 
  
  
  
  
  19.        try { return m.put(key, value); }  
    20. 
  
  
  
  
  20.        finally { w.unlock(); }  
    21. 
  
  
  
  
  21.    }  
    22. 
  
  
  
  
  22.    public void clear() {  
    23. 
  
  
  
  
  23.        w.lock();  
    24. 
  
  
  
  
  24.        try { m.clear(); }  
    25. 
  
  
  
  
  25.        finally { w.unlock(); }  
    26. 
  
  
  
  
  26.    }  
    27. 
  
  
  
  
  27. }  
    28. 
  
  
  
  
  28.  
    29.

要记得write锁是独占的，它一样可以使用ReentrantLock的Condition功能。

使用任何的锁都要通过try catch 或者 finally 来处理异常，避免忘记unlock。

#p#

4. 同步辅助类

你提交了一些任务，但你想等它们都完成了再做另外一些事情;你提交了一些任务，但是不想让它们立刻执行，等你喊123开始的时候，它们才开始执行;等等这些场景，线程之间需要相互配合，或者等待某一个条件成熟执行。这些场景想你就需要用到同步辅助类。

4.1 CountDownLatch

CountDownLatch 内部有个计数器，通过构造函数来指定。这个类就好比是倒计时的电子牌，当倒计时为0的时候就可以一起做一些事情。

摘自JavaDoc的方法介绍

摘自JavaDoc的例子

 
 
 
 
    
  
  
  
  
  1. class Driver { // ...  
    2. 
  
  
  
  
  2.   void main() throws InterruptedException {  
    3. 
  
  
  
  
  3.     CountDownLatch startSignal = new CountDownLatch(1);  
    4. 
  
  
  
  
  4.     CountDownLatch doneSignal = new CountDownLatch(N);  
    5. 
  
  
  
  
  5.  
    6. 
  
  
  
  
  6.     for (int i = 0; i < N; ++i) // create and start threads  
    7. 
  
  
  
  
  7.       new Thread(new Worker(startSignal, doneSignal)).start();  
    8. 
  
  
  
  
  8.  
    9. 
  
  
  
  
  9.     doSomethingElse();            // don't let run yet  
    10. 
  
  
  
  
  10.     startSignal.countDown();      // let all threads proceed  
    11. 
  
  
  
  
  11.     doSomethingElse();  
    12. 
  
  
  
  
  12.     doneSignal.await();           // wait for all to finish  
    13. 
  
  
  
  
  13.   }  
    14. 
  
  
  
  
  14. }  
    15. 
  
  
  
  
  15.  
    16. 
  
  
  
  
  16. class Worker implements Runnable {  
    17. 
  
  
  
  
  17.   private final CountDownLatch startSignal;  
    18. 
  
  
  
  
  18.   private final CountDownLatch doneSignal;  
    19. 
  
  
  
  
  19.   Worker(CountDownLatch startSignal, CountDownLatch doneSignal) {  
    20. 
  
  
  
  
  20.      this.startSignal = startSignal;  
    21. 
  
  
  
  
  21.      this.doneSignal = doneSignal;  
    22. 
  
  
  
  
  22.   }  
    23. 
  
  
  
  
  23.   public void run() {  
    24. 
  
  
  
  
  24.      try {  
    25. 
  
  
  
  
  25.        startSignal.await();  
    26. 
  
  
  
  
  26.        doWork();  
    27. 
  
  
  
  
  27.        doneSignal.countDown();  
    28. 
  
  
  
  
  28. } catch (InterruptedException ex) {} // return;  
    29. 
  
  
  
  
  29.   }  
    30. 
  
  
  
  
  30.  
    31. 
  
  
  
  
  31.   void doWork() { ... }  
    32. 
  
  
  
  
  32. }  
    33. 
  
  
  
  
  33.  
    34.

当CountDownLatch(1)的时候，它就好比是个信号枪了。

4.2 CyclicBarrier

 
 
 
 
    
  
  
  
  
  1. new CyclicBarrier(N,   
    2. 
  
  
  
  
  2.             new Runnable() {  
    3. 
  
  
  
  
  3.               public void run() {   
    4. 
  
  
  
  
  4.                mergeRows(...);   
    5. 
  
  
  
  
  5.     }  
    6. 
  
  
  
  
  6. }); 
    7.

这个同步辅助类，它让多个线程可以在多个屏障点进行等待，所以叫cyclic，而且有个附加选择你可以在线程到达屏障点后执行一个任务(在释放其他线程之前)

为了帮助你理解，假设一个场景。

有一个任务，A、B、C分别从三个仓库(甲乙丙)搬运不同3个不同的零件到客户X的公司，然后再一起组装机器，完成后一起坐车去公司总部。

这个任务需要ABC三个线程同时进行，但是由于从仓库到客户X那边距离不等、交通状态未知的情况下，所花费的时间也不等。同时由于三个人负责的零件不同，所以安装机器的时候花费时间也不一样。这个场景中有两个需要线程间等待的地方。CyclicBarrier就可以闪亮登场了。

 
 
 
 
    
  
  
  
  
  1. public class Main3 {  
    2. 
  
  
  
  
  2.  
    3. 
  
  
  
  
  3. public static void main(String[] args) {  
    4. 
  
  
  
  
  4.  
    5. 
  
  
  
  
  5. CyclicBarrier barrier = new CyclicBarrier(3,new Runnable() {  
    6. 
  
  
  
  
  6.  
    7. 
  
  
  
  
  7. @Override 
    8. 
  
  
  
  
  8.  
    9. 
  
  
  
  
  9. public void run() {  
    10. 
  
  
  
  
  10.  
    11. 
  
  
  
  
  11. System.out.println("到达公共屏障点");  
    12. 
  
  
  
  
  12.  
    13. 
  
  
  
  
  13. }  
    14. 
  
  
  
  
  14.  
    15. 
  
  
  
  
  15. });  
    16. 
  
  
  
  
  16.  
    17. 
  
  
  
  
  17. ExecutorService es = Executors.newCachedThreadPool();  
    18. 
  
  
  
  
  18.  
    19. 
  
  
  
  
  19. es.submit(new Worker("A", 5000, 8000, barrier));  
    20. 
  
  
  
  
  20.  
    21. 
  
  
  
  
  21. es.submit(new Worker("B", 2000, 16000, barrier));  
    22. 
  
  
  
  
  22.  
    23. 
  
  
  
  
  23. es.submit(new Worker("C", 9000, 2000, barrier));  
    24. 
  
  
  
  
  24.  
    25. 
  
  
  
  
  25. es.shutdown();  
    26. 
  
  
  
  
  26.  
    27. 
  
  
  
  
  27. }  
    28. 
  
  
  
  
  28.  
    29. 
  
  
  
  
  29. static class Worker implements Runnable {  
    30. 
  
  
  
  
  30.  
    31. 
  
  
  
  
  31. String name;  
    32. 
  
  
  
  
  32.  
    33. 
  
  
  
  
  33. int t1;// 搬运零件所需要的时间  
    34. 
  
  
  
  
  34.  
    35. 
  
  
  
  
  35. int t2;// 参与组装工作需要的时间  
    36. 
  
  
  
  
  36.  
    37. 
  
  
  
  
  37. CyclicBarrier barrier;  
    38. 
  
  
  
  
  38.  
    39. 
  
  
  
  
  39. public Worker(String name, int t1, int t2, CyclicBarrier barrier) {  
    40. 
  
  
  
  
  40.  
    41. 
  
  
  
  
  41. super();  
    42. 
  
  
  
  
  42.  
    43. 
  
  
  
  
  43. this.name = name;  
    44. 
  
  
  
  
  44.  
    45. 
  
  
  
  
  45. this.t1 = t1;  
    46. 
  
  
  
  
  46.  
    47. 
  
  
  
  
  47. this.t2 = t2;  
    48. 
  
  
  
  
  48.  
    49. 
  
  
  
  
  49. this.barrier = barrier;  
    50. 
  
  
  
  
  50.  
    51. 
  
  
  
  
  51. }  
    52. 
  
  
  
  
  52.  
    53. 
  
  
  
  
  53. @Override 
    54. 
  
  
  
  
  54.  
    55. 
  
  
  
  
  55. public void run() {  
    56. 
  
  
  
  
  56.  
    57. 
  
  
  
  
  57. try {  
    58. 
  
  
  
  
  58.  
    59. 
  
  
  
  
  59. print(name + " 开始搬运零件");  
    60. 
  
  
  
  
  60.  
    61. 
  
  
  
  
  61. Thread.sleep(t1);// 模拟搬运时间  
    62. 
  
  
  
  
  62.  
    63. 
  
  
  
  
  63. print(name + " 到达目的地");  
    64. 
  
  
  
  
  64.  
    65. 
  
  
  
  
  65. int a = barrier.await(); // 等待其他人  
    66. 
  
  
  
  
  66.  
    67. 
  
  
  
  
  67. if(a==0){  
    68. 
  
  
  
  
  68.  
    69. 
  
  
  
  
  69. //说明是最后一个到的可以执行特殊操作  
    70. 
  
  
  
  
  70.  
    71. 
  
  
  
  
  71. }  
    72. 
  
  
  
  
  72.  
    73. 
  
  
  
  
  73. print(name + " 开始组装机器");  
    74. 
  
  
  
  
  74.  
    75. 
  
  
  
  
  75. Thread.sleep(t2);// 模拟组装时间.  
    76. 
  
  
  
  
  76.  
    77. 
  
  
  
  
  77. print(name + " 完成组装机器");  
    78. 
  
  
  
  
  78.  
    79. 
  
  
  
  
  79. barrier.await(); // 等待其他人组装完毕  
    80. 
  
  
  
  
  80.  
    81. 
  
  
  
  
  81. print(name + " 一起回总公司");  
    82. 
  
  
  
  
  82.  
    83. 
  
  
  
  
  83. } catch (Exception e) {  
    84. 
  
  
  
  
  84.  
    85. 
  
  
  
  
  85. e.printStackTrace();  
    86. 
  
  
  
  
  86.  
    87. 
  
  
  
  
  87. }  
    88. 
  
  
  
  
  88.  
    89. 
  
  
  
  
  89. }  
    90. 
  
  
  
  
  90.  
    91. 
  
  
  
  
  91. }  
    92. 
  
  
  
  
  92.  
    93. 
  
  
  
  
  93. static SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss");  
    94. 
  
  
  
  
  94.  
    95. 
  
  
  
  
  95. static void print(String x) {  
    96. 
  
  
  
  
  96.  
    97. 
  
  
  
  
  97. System.out.println( sdf.format(new Date()) + ": "+x);  
    98. 
  
  
  
  
  98.  
    99. 
  
  
  
  
  99. }  
    100. 
  
  
  
  
  100.  
    101. 
  
  
  
  
  101. }  
    102.

4.3 Semaphore

一个经典的信号量计数器。一般被用来控制对共享资源同时访问线程数量的控制。

特殊情况下信号量设置为1，那么就类似互斥锁的功能。

此类的构造方法可选地接受一个公平 参数。当设置为 false 时，此类不对线程获取锁的顺序做任何保证。和之前提到的争用获取顺序一样，在非公平模式下，系统将获得更好的吞吐量，jvm也会保证在非公平模式下让所有线程得到访问机会。

【编辑推荐】

  1. java.util.concurrent 您不知道的 5 件事
  2. util.concurrent移植到C#

文章标题：ConcurrentInJava
URL地址：http://www.shufengxianlan.com/qtweb/news37/66087.html

网站建设、网络推广公司-创新互联，是专注品牌与效果的网站制作，网络营销seo公司；服务项目有等

广告

声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源： 创新互联

猜你还喜欢下面的内容

网站营销知识

同城分类信息

响应式网站设计　　 　眉山电信机房　　 　航龙航空　　 　手机网站建设公司　　 　成都二枢电信服务器托管　　 　证企宝企业　　 　高防机房　　 　鸿艺文化　　 　子晨木包装箱　　 　网站排名　　 　高县网站建设　　 　手机网站制作　　 　定制网站开发　　 　蜀锦在线　　 　重庆水土三线托管　　 　app开发公司　　 　成都茶叶销售　　 　重庆企业网站建设　　 　域名注册　　 　高端网站设计