第6章-Java并发容器和框架

发表于 | 分类于 |

本章主要介绍了并发容器（ConcurrentHashMap、ConcurrentLinkedQueue、阻塞队列）和并发框架（Fork/Join框架），并分析了其内部原理。

ConcurrentHashMap的实现原理与使用

ConcurrentHashMap是线程安全且高效的HashMap。

为什么要使用ConcurrentHashMap

  • 线程不安全的HashMap
  • 效率低下的HashTable
  • ConcurrentHashMap的锁分段技术可有效提升并发访问率

ConcurrentHashMap的结构

ConcurrentHashMap是由Segment数组结构和HashEntry数组结构组成。Segment是一种可重入锁（ReentrantLock），在ConcurrentHashMap里扮演锁的角色；HashEntry则用于存储键值对数据。
核心操作：

  • get操作
  • put操作
    • 是否需要扩容
    • 如何扩容
  • size操作

ConcurrentHashMap的类图：
Alt text
ConcurrentHashMap的结构图：
Alt text

ConcurrentLinkedQueue

非阻塞的方式来实现线程安全队列。

ConcurrentLinkedQueue的结构

ConcurrentLinkedQueue的类图：
Alt text

入队列

入队列就是将入队节点添加到队列的尾部。

[源码]
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
public boolean offer(E e) {
	if (e == null) throw new NullPointerException();
	// 入队前，创建一个入队节点
	Node<E> n = new Node<E>(e);
	retry:
	// 死循环，入队不成功反复入队。
	for (;;) {
		// 创建一个指向tail节点的引用
		Node<E> t = tail;
		// p用来表示队列的尾节点，默认情况下等于tail节点。
		Node<E> p = t;
		for (int hops = 0; ; hops++) {
			// 获得p节点的下一个节点。
			Node<E> next = succ(p);
			// next节点不为空，说明p不是尾节点，需要更新p后在将它指向next节点
			if (next != null) {
				// 循环了两次及其以上，并且当前节点还是不等于尾节点
				if (hops > HOPS && t != tail)
					continue retry;
				p = next;
			}
			// 如果p是尾节点，则设置p节点的next节点为入队节点。
			else if (p.casNext(null, n)) {
				/*如果tail节点有大于等于1个next节点，则将入队节点设置成tail节点，
				更新失败了也没关系，因为失败了表示有其他线程成功更新了tail节点*/
				if (hops >= HOPS)
					casTail(t, n); // 更新tail节点，允许失败
				return true;
			}
			// p有next节点,表示p的next节点是尾节点，则重新设置p节点
			else {
				p = succ(p);
			}
		}
	}
}

出队列

出队列的就是从队列里返回一个节点元素，并清空该节点对元素的引用。

[源码]
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
public E poll() {
	Node<E> h = head;
	// p表示头节点，需要出队的节点
	Node<E> p = h;
	for (int hops = 0;; hops++) {
		// 获取p节点的元素
		E item = p.getItem();
		// 如果p节点的元素不为空，使用CAS设置p节点引用的元素为null,
		// 如果成功则返回p节点的元素。
		if (item != null && p.casItem(item, null)) {
			if (hops >= HOPS) {
				// 将p节点下一个节点设置成head节点
				Node<E> q = p.getNext();
				updateHead(h, (q != null) q : p);
			}
			return item;
		}
		// 如果头节点的元素为空或头节点发生了变化，这说明头节点已经被另外
		// 一个线程修改了。那么获取p节点的下一个节点
		Node<E> next = succ(p);
		// 如果p的下一个节点也为空，说明这个队列已经空了
		if (next == null) {
			// 更新头节点。
			updateHead(h, p);
			break;
		}
		// 如果下一个元素不为空，则将头节点的下一个节点设置成头节点
		p = next;
	}
	return null;
}

Java中的阻塞队列

本节将介绍什么是阻塞队列，以及Java中阻塞队列的4种处理方式，并介绍Java 7中提供的7种阻塞队列，最后分析阻塞队列的一种实现方式。

什么是阻塞队列

阻塞队列（BlockingQueue）是一个支持两个附加操作的队列。
阻塞队列的4种处理方式：
Alt text

Java里的阻塞队列

JDK 7提供了7个阻塞队列，如下：

  • ArrayBlockingQueue：一个由数组结构组成的有界阻塞队列。
  • LinkedBlockingQueue：一个由链表结构组成的有界阻塞队列。
  • PriorityBlockingQueue：一个支持优先级排序的无界阻塞队列。
  • DelayQueue：一个使用优先级队列实现的无界阻塞队列。
  • SynchronousQueue：一个不存储元素的阻塞队列。
  • LinkedTransferQueue：一个由链表结构组成的无界阻塞队列。
  • LinkedBlockingDeque：一个由链表结构组成的双向阻塞队列。

ArrayBlockingQueue

ArrayBlockingQueue是一个用数组实现的有界阻塞队列。

[源码]
1
2
3
4
5
6
7
8
9
ArrayBlockingQueue fairQueue = new ArrayBlockingQueue(1000,true);
public ArrayBlockingQueue(int capacity, boolean fair) {
	if (capacity <= 0)
		throw new IllegalArgumentException();
	this.items = new Object[capacity];
	lock = new ReentrantLock(fair);
	notEmpty = lock.newCondition();
	notFull = lock.newCondition();
}

…..

阻塞队列的实现原理

使用通知模式实现。底层是Condition来实现。

[源码]
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
private final Condition notFull;
private final Condition notEmpty;
public ArrayBlockingQueue(int capacity, boolean fair) {
	// 省略其他代码
	notEmpty = lock.newCondition();
	notFull = lock.newCondition();
}
public void put(E e) throws InterruptedException {
	checkNotNull(e);
	final ReentrantLock lock = this.lock;
	lock.lockInterruptibly();
	try {
		while (count == items.length)
			notFull.await();
		insert(e);
	} finally {
		lock.unlock();
	}
} 
public E take() throws InterruptedException {
	final ReentrantLock lock = this.lock;
	lock.lockInterruptibly();
	try {
		while (count == 0)
			notEmpty.await();
		return extract();
	} finally {
		lock.unlock();
	}
} 
private void insert(E x) {
	items[putIndex] = x;
	putIndex = inc(putIndex);
	++count;
	notEmpty.signal();
}

Fork/Join框架

本节将会介绍Fork/Join框架的基本原理、算法、设计方式、应用与实现等。

什么是Fork/Join框架

Fork/Join框架是Java 7提供的一个用于并行执行任务的框架，是一个把大任务分割成若干
个小任务，最终汇总每个小任务结果后得到大任务结果的框架。

流程图：
Alt text

使用Fork/Join框架

让我们通过一个简单的需求来使用Fork/Join框架，需求是：计算1+2+3+4的结果。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.Future;
import java.util.concurrent.RecursiveTask;
public class CountTask extends RecursiveTask<Integer> {
	private static final int THRESHOLD = 2;　　// 阈值
	private int start;
	private int end;
	public CountTask(int start, int end) {
		this.start = start;
		this.end = end;
	}
	@Override
	protected Integer compute() {
		int sum = 0;
		// 如果任务足够小就计算任务
		boolean canCompute = (end - start) <= THRESHOLD;
		if (canCompute) {
			for (int i = start; i <= end; i++) {
				sum += i;
			}
		} else {
			// 如果任务大于阈值，就分裂成两个子任务计算
			int middle = (start + end) / 2;
			CountTask leftTask = new CountTask(start, middle);
			CountTask rightTask = new CountTask(middle + 1, end);
			// 执行子任务
			leftTask.fork();
			rightTask.fork();
			// 等待子任务执行完，并得到其结果
			int leftResult=leftTask.join();
			int rightResult=rightTask.join();
			// 合并子任务
			sum = leftResult + rightResult;
		}
		return sum;
	}
	public static void main(String[] args) {
		ForkJoinPool forkJoinPool = new ForkJoinPool();
		// 生成一个计算任务，负责计算1+2+3+4
		CountTask task = new CountTask(1, 4);
		// 执行一个任务
		Future<Integer> result = forkJoinPool.submit(task);
		try {
			System.out.println(result.get());
		} catch (InterruptedException e) {
		} catch (ExecutionException e) {
		}
	}
}

Fork/Join框架的异常处理

ForkJoinTask提供了isCompletedAbnormally()方法来检查任务是否已经抛出异常或已经被取消了，并且可以通过ForkJoinTask的getException方法获取异常。getException方法返回Throwable对象，如果任务被取消了则返回CancellationException。如果任务没有完成或者没有抛出异常则返回null。使用如下代码：

1
2
3
4
if(task.isCompletedAbnormally())
{
	System.out.println(task.getException());
}