Java Concurrency

1. 简介

1.1 多线程的好处

• 提高性能，提高吞吐量，开发多核CPU的性能;
• 使UI应答更顺畅

1.2 多线程的坏处

• 安全：多线程环境下，未同步的操作，由于编译器的优化，硬件，运行时库的优化，可能导致执行顺序不可预期，共享数据状态不可预期，共享数据的变更对其他线程不可见;
• 死锁：多线程环境下，不恰当的锁操作可能导致死锁;
• 性能：过多的线程可能导致性能下降，因为需要线程调度，切换上下文，加锁释放锁，对共享数据内存刷新等;

2. 线程安全

2.1 几个概念

2.1.1 什么是线程安全

类调用的代码被多个线程同时访问，被CPU调度而暂停执行，或恢复，不需要额外的同步或协调，总能保持一致的状态。

 

2.1.2 原子性

不可分割性的操作（CPU指令、操作系统指令、VM指令）

 

2.1.3 线程

系统对操作进行调度的最小单元（操作系统、VM）

 

2.1.4 并发

当多于一个线程（进程）同时争抢同一个资源时。资源包括执行程序，访问IO，访问内存数据块

2.1.5 阻塞

线程发生的被动式等待行为（如等待获取独占性资源、等待结果等）。

2.1.6 同步与异步

同步：串行化处理

异步：并行化处理

2.1.7 线程状态转换

如下图：

注意：yield()和sleep()并不会释放锁

 

 

3. 共享对象

3.1 可见性

由于JVM读写64位的long或double变量，是两次32位的操作，如果变量没有volatile修饰符，多个线程读写可能导致读写的高位和地位出错。

3.1.4 volatile修饰符

volatile是一个较弱的同步修饰符，JVM保证对它修饰变量的更新对其他线程可见，编译器和运行时库对这个变量不会做优化，它的值不会放在CPU寄存器或缓存中（这里的变量对其他线程不可见）。

但我们不能太严重的依赖这个修饰符，它能提供的同步很脆弱，它只能保证可见，而锁既保证了可见又保证了操作的原子性;

最佳实践是：

• 写入该变量不依赖它的当前值，或只有一个线程更新该变量;

• 生命周期事件的标志（例如：初始化或结束）。

3.1.5 可重入

可重入（当一个线程请求其他线程已占有的锁时，请求线程将会被阻塞，然而内部锁是可重入的，线程在试图获得它自己占有的锁时，请求将会成功）

3.2 ThreadLocal对象

一个线程私有的存储空间，主要用来存储一个上下文对象。

第一次调用该对象的get方法时，initialValue会被调用作为初始化。ThreadLocal对象存储在Thread对象中，当线程终止，ThreadLocal对象也就可以回收了。

注意：线程池中的线程可能会循环利用，因此其附加的ThreadLocal对象需及时清理，否则容易造成内存泄漏和脏数据（上次ThreadLocal的数据）

3.3 不可变对象（immutable）

对象创建之后状态不能更改，所有域都是final。不可变对象是线程安全的。

4.同步的集合（Synchronized Collections）

同步的集合封装了集合的状态且同步了所有的public方法, 同时只能有一个线程访问该集合。

Collections.synchronizedXxx返回同步的集合。

4.1 遍历和并发异常（ConcurrentModificationException）

同步对象在遍历（iterator或for）时，如果有其他线程修改了该对象，则会抛出并发异常（ConcurrentModificationException），即fail-fast。

5.并发集合（Concurrent Collections）

并发集合替换同步的集合会显著的提高性能，但牺牲了较少（较少涉及）的线程安全。

5.1 CurrentHashMap

CurrentHashMap是设计来替代Collections.synchronizedMap(new HashMap())

CurrentHashMap是分段锁，且部分方法（如：size, isEmpty，被认为在多线程环境下不是很有用，或者用的很少)未加锁，且遍历时不会抛异常（ConcurrentModificationException）（因此遍历时可能得到与获得迭代器时不同的元素，被认为是可以容忍的），可以多个线程同时访问。

5.2 CopyOnWriteArrayList和CopyOnWriteArraySet

CopyOnWriteArrayList则是Collections.synchronizedList(new List())的替代品，同样，CopyOnWriteArraySet则是Collections.synchronizedSet(new Set())。

由于是写时拷贝的机制，因此适合遍历远多于修改的情况。

5.3 阻塞队列（Blocking Queue）

阻塞队列特别适合生产者消费者模式。

其实现有：

• LinkedBlockingQueue，ArrayBlockingQueue，分别对应：LinkedList，ArrayList，但比Collections.synchronizedList()有更好的并发性，他们都是FIFO队列。

• PriorityBlockingQueue，是优先级排序的队列。

• SynchronousQueue，不是一个真正意义上的队列，它不占用存储空间，它维护一个等待入队和出队的线程列表，它直接将元素递交给处理线程，put和take方法都会阻塞直到有线程处理，因此一般适合有足够的消费者处理的情况。

Java6加入了Deque和BlockingDeque，分别扩展了Queue和BlockingQueue。Deque是一个两端都可以插入，删除元素的队列。实现有ArrayDeque和LinkedBlockingDeque。

5.4 同步器

5.4.1 CountDownLatch

CountDownLatch像一个大门，它初始化一个计数，在计数没减少到0之前，所有调用CountDownLatch.await()的线程都等待，到0时，便释放所有的线程，此时CountDownLatch的状态为终止，门也就一直打开了。

适合于线程等待一个初始化条件；等待所有参与者的加入；调度线程等待工作线程的完成。

5.4.2 FutureTask

等待一个FutureTask完成并返回其结果，FutureTask.get()会阻塞直到FutureTask到完成状态，然后返回结果，或者抛出包装的异常（ExecutionException）。

5.4.3 Semaphores（信号量）

Semaphore是一个资源池的概念，构造函数初始化池的大小，acquire获取一个资源（没资源的时候可能被阻塞），release放回一个资源。

5.4.4 CyclicBarrier

允许一组线程相互等待，直到都达到某个屏障点，然后释放全部线程，该CycilcBarrier还可以用，此过程可以循环。

CyclicBarrier构造函数支持一个Runnable对象（barrier action ）作为参数，当所有线程达到某个屏障点，但尚未释放前在一个子线程中执行，通常用来更新共享的状态。

CyclicBarrier采用快速失败（fast-fail）和荣辱与共（all-or-none）策略，线程由于异常，中断（interrupt），重置或超时离开屏障点，将导致其他线程都抛出BrokenBarrierException或InterruptedException。

CyclicBarrier.await()方法将返回一个标识线程到达屏障点的索引，0表示最后一个到达的，getParties() - 1表示最先到的。

CountDownLatch和CyclicBarrier的例子：

package org.jamie.demo;

import java.util.Random;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;

public class CyclicBarrierDemo {

	public static class Worker extends Thread {
		private CyclicBarrier barrier;
		private CountDownLatch latch;

		public Worker(CyclicBarrier barrier, CountDownLatch latch) {
			super();
			this.barrier = barrier;
			this.latch = latch;
		}

		@Override
		public void run() {
			System.out.println(this + " is going to company...");
			Random ran = new Random();
			try {
				Thread.sleep(ran.nextInt(3000));
			} catch (InterruptedException e1) {
				e1.printStackTrace();return;
			}
			try {
				int index = barrier.await(); // wait other workers
				System.out.println(this + " is the " + (barrier.getParties() - index)  + " one got to company");
			} catch (InterruptedException e) {
				e.printStackTrace(); return;
			} catch (BrokenBarrierException e) {
				e.printStackTrace(); return;
			}

			// working
			try {
				Thread.sleep(ran.nextInt(10000));
			} catch (InterruptedException e1) {
				e1.printStackTrace();return;
			}
			try {
				int index = barrier.await(); // wait other workers
				System.out.println(this + " is the " + (barrier.getParties() - index)  + " one finished job");
			} catch (InterruptedException e) {
				e.printStackTrace(); return;
			} catch (BrokenBarrierException e) {
				e.printStackTrace(); return;
			}
			latch.countDown();
		}
	}

	public static void main(String[] args) throws InterruptedException {
		int workerCount = 5;
		CountDownLatch latch = new CountDownLatch(5);
		CyclicBarrier barrier = new CyclicBarrier(workerCount, new Runnable() {

			@Override
			public void run() {
				System.out.println(Thread.currentThread() + " arrive barrier");
			}
			
		});

		for (int i = 0; i < workerCount; ++i) {
			new Worker(barrier, latch).start();
		}

		latch.await();// wait all worker finished
	}
}

  输出：

写道

Thread[Thread-0,5,main] is going to company...
Thread[Thread-2,5,main] is going to company...
Thread[Thread-1,5,main] is going to company...
Thread[Thread-4,5,main] is going to company...
Thread[Thread-3,5,main] is going to company...
Thread[Thread-2,5,main] arrive barrier
Thread[Thread-2,5,main] is the 5 one got to company
Thread[Thread-3,5,main] is the 1 one got to company
Thread[Thread-4,5,main] is the 2 one got to company
Thread[Thread-0,5,main] is the 3 one got to company
Thread[Thread-1,5,main] is the 4 one got to company
Thread[Thread-1,5,main] arrive barrier
Thread[Thread-1,5,main] is the 5 one finished job
Thread[Thread-3,5,main] is the 1 one finished job
Thread[Thread-0,5,main] is the 2 one finished job
Thread[Thread-4,5,main] is the 4 one finished job
Thread[Thread-2,5,main] is the 3 one finished job