第10章 常用的并发集合
2.1 ConcurrentHashMap
2.1.1 ConcurrentHashMap介绍
为什么要学ConcurrentHashMap？
HashMap是线程不安全的。
也有一些线程安全的映射表，如下：
早期java版本中有HashTable。
由Collections包装的集合类，比如：Collections.SynchronizedMap，在Collections.SynchronizedMap类中对Map的方法进行包装，方法内部使用synchronized修饰，经过包装实现了map的线程安全。
ConcurrentHashMap是JUC包中提供的线程安全的映射表，它的性能相比其它映射表更高。
2.1.2 ConcurrentHashMap源码解析
ConcurrentHashMap比HashTable在性能上有很大的提升，HashTable是对全表进行锁定，ConcurrentHashMap则是局部锁定。ConcurrentHashMap在jdk1.7、1.8上设计不一样，1.7在数组+链表的基础上采用分段锁技术保证线程安全，1.8在数组+链表+红黑树的基础上采用CAS+synchronized保证线程安全。
ConcurrentHashMap实现Map接口，针对哈希表的一些常用方法这里不再测试，下边以jdk1.8为准通过阅读源码理解下它的运行原理：
1、put方法
public V put(K key, V value) {
return putVal(key, value, false);
}
/** Implementation for put and putIfAbsent */
//onlyIfAbsent: false每次用新值覆盖旧值，true：已存在相同key不作任何操作
final V putVal(K key, V value, boolean onlyIfAbsent) {
if (key == null || value == null) throw new NullPointerException();
//计算机哈希值并保证正数
//(h ^ (h >>> 16)) & HASH_BITS (h ^ (h >>> 16))是hashMap中计算哈希的方法，& HASH_BITS的作用是将负数转正数 HASH_BITS是31位1
int hash = spread(key.hashCode());
int binCount = 0;
for (Node<K,V>[] tab = table;;) {
Node<K,V> f; int n, i, fh;
if (tab == null || (n = tab.length) == 0)
//使用cas初始化table
tab = initTable();
//桶的位置为空则用cas方式创建头结点，i是桶的位置
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
if (casTabAt(tab, i, null,
new Node<K,V>(hash, key, value, null)))
break; // no lock when adding to empty bin
}
else if ((fh = f.hash) == MOVED)//如果发现有线程在扩容则先协助扩容
tab = helpTransfer(tab, f);
else {//哈希冲突
V oldVal = null;
//使用同步锁对头结点进行加锁
synchronized (f) {
if (tabAt(tab, i) == f) {//如果头结点没有被移动过
if (fh >= 0) {//普通结点
binCount = 1;
for (Node<K,V> e = f;; ++binCount) {
K ek;
//根据key查找，如果已经存在
if (e.hash == hash &&
((ek = e.key) == key ||
(ek != null && key.equals(ek)))) {
oldVal = e.val;
//更新旧值
if (!onlyIfAbsent)
e.val = value;
break;
}
Node<K,V> pred = e;
//如果结点不存在则新增一个结点到链表尾部
if ((e = e.next) == null) {
pred.next = new Node<K,V>(hash, key,
value, null);
break;
}
}
}
else if (f instanceof TreeBin) {//如果是红黑树结点
Node<K,V> p;
binCount = 2;
//查找key是否在红黑树中，如果没有则添加
if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
value)) != null) {
oldVal = p.val;
//更新旧值
if (!onlyIfAbsent)
p.val = value;
}
}
}
}
if (binCount != 0) {
//如是链表长度大于阀值则将链表转成红黑树
if (binCount >= TREEIFY_THRESHOLD)
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
}
}
addCount(1L, binCount);//统计结点个数，检查是否需要扩容
return null;
}initTable()
private final Node<K,V>[] initTable() {
Node<K,V>[] tab; int sc;
while ((tab = table) == null || tab.length == 0) {
//sizeCtl用于控制初始化和扩容
//如果sizeCtl小于0说明有线程执行CAS成功正在初始化，此时当前线程让出CPU
if ((sc = sizeCtl) < 0)
Thread.yield(); // lost initialization race; just spin
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
try {
if ((tab = table) == null || tab.length == 0) {
//默认大小16
int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
table = tab = nt;
sc = n - (n >>> 2);
}
} finally {
sizeCtl = sc;
}
break;
}
}
return tab;
}tabAt:
static final <K,V> Node<K,V> tabAt(Node<K,V>[] tab, int i) {
//直接获取指定内存数据，保证每次拿到的就是最新数据
return (Node<K,V>)U.getObjectVolatile(tab, ((long)i << ASHIFT) + ABASE);
}2、get方法
get方法全程没有加锁，性能很高，源代码如下：
public V get(Object key) {
Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
//计算机哈希值并保证正数
int h = spread(key.hashCode());
//如果表的长度大于0开始定位桶
if ((tab = table) != null && (n = tab.length) > 0 &&
//(n - 1) & h)桶的下标
(e = tabAt(tab, (n - 1) & h)) != null) {
if ((eh = e.hash) == h) {//如果桶位置上的结点的哈希与查找的key的哈希相同则开始比较key是否相等，如果相等则返回key对应 的value
if ((ek = e.key) == key || (ek != null && key.equals(ek)))
return e.val;
}
else if (eh < 0)//如果eh为负数则表示链表在扩容中或为红黑树，调用find查找key
return (p = e.find(h, key)) != null ? p.val : null;
//以上两种情况都不符合则开始遍历每个结点
while ((e = e.next) != null) {
if (e.hash == h &&
((ek = e.key) == key || (ek != null && key.equals(ek))))
return e.val;
}
}
return null;
}2.1.3 ConcurrentHashMap案例
统计字符串每个字符出现的次数。
1、首先准备等待统计的数据，将0到9的10个数字循环100遍放入集合。
2、首先用HashMap实现，创建一个HashMap对象，遍历集合中等待统计的字符，如果字符在map中存在则计数加1，否则初始为1。
代码如下：
package com.yjoffer.javase.thread.collection;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
/**
- 线程安全集合类测试
- @author 预见猿份(www.yjoffer.com)
- @version 1.0
*/
public class ConcurrentHashMapTest {//准备等待统计的数据，统一放入集合
private static List data_init(){
String data = "0123456789";
List<String> characters = new ArrayList<>();
for (int j = 0; j < 100; j++) {
for (int i = 0; i < data.length(); i++) {
char c = data.charAt(i);
characters.add(String.valueOf(c));
}
}
return characters;按预期map中共有10个key，每个key对应的value为100。
执行程序，存在错误。
下边用线程安全的集合类ConcurrentHashMap进行修改。
首先将new HashMap()更改为new ConcurrentHashMap()，再次运行程序，结果 仍然 存在错误。
下边分析错误，导致问题的代码如下：
Integer integer = map.get(c);
if(integer == null){
map.put(c,1);
}else{
map.put(c,integer+1);
}虽然get方法是线程安全，put方法是线程安全，但是由于是多线程执行存在线程切换，上边整体代码线程不安全。
修改方案1：使用synchronized将上边的代码保持原子性。
synchronized (map){
Integer integer = map.get(c);
if(integer == null){
map.put(c,1);
}else{
map.put(c,integer+1);
}
}运行程序，结果正确。
此方法是将整个表对象作为锁对象，性能较低。
修改方案2：使用CAS方法
下边使用putIfAbsent方法实现key判断的逻辑，如果key不存在则用一个初始值填充，这里用原子类填充，并利用原子类的CAS方法自增1。
代码修改如下：
AtomicLong atomicLong = map.putIfAbsent(c, new AtomicLong(1));
if(atomicLong!=null){
atomicLong.incrementAndGet();
}map.putIfAbsent(c, new AtomicLong(1));代码相当于下边的代码：
if (!map.containsKey(key)) return map.put(key, new AtomicLong(1)); else return map.get(key);完整代码如下：
public static void test_concurrentMap(){
List<String> characters = data_init();
Map<String, AtomicLong> map = new ConcurrentHashMap<>();
ExecutorService executorService = Executors.newCachedThreadPool();
characters.stream().forEach(c->{
executorService.execute(()->{
AtomicLong atomicLong = map.putIfAbsent(c, new AtomicLong(1));
if(atomicLong!=null){
atomicLong.incrementAndGet();
}
});
});
shutdown(executorService);
System.out.println(map.size());
System.out.println(map);
}2.2 ArrayBlockingQueue
2.2.1 阻塞队列介绍
在生产者与消费者模式中，阻塞队列是生产者与消费者的中介，它可以很好的将生产者与消费者解耦合，提高软件 的健壮性和可扩展性。
下图中的共享数据容器可用阻塞队列实现。
BlockingQueue是阻塞队列的接口，它的常用实现类如下：
1、ArrayBlockingQueue，有边界的阻塞队列，内部使用数组实现。
2、DelayQueue，无界阻塞队列，指定每个元素的到期时长，元素在延迟到期时才被使用。
3、LinkedBlockingQueue，无界阻塞队列（也可有界），内部使用链表实现。
4、PriorityBlockingQueue，无界阻塞队列，指定每个元素的优先级，按优先级排序。
5、SynchronousQueue，具有一个元素的阻塞队列，每个插入操作必须等待另一个线程相应的删除操作，反之亦然。
6、LinkedBlockingDeque，由链表组成的双向阻塞队列
7、LinkedTransferQueue，由链表组成的无界阻塞TransferQueue队列2.2.2 ArrayBlockingQueue源码分析
1、数组结构
ArrayBlockingQueue是一个基于数组实现的阻塞队列，它是有界队列所以在构造时需要 指定队列的大小，它的常用方法有add ,offer,put，remove,poll,take,peek，在使用ArrayBlockingQueue时注意哪些是阻塞方法哪些不是阻塞方法，比如：add、offer不是阻塞方法，put是阻塞方法。
ArrayBlockingQueue实现了队列、阻塞队列、集合的方法，关于出队和入队的基本操作这里不再演示，下边通过源代码分析它的原理。
ArrayBlockingQueue使用数组存储数据，它的重要成员如下：
public class ArrayBlockingQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable {
...
//底层使用数组存储元素
final Object[] items;
//队头下标
int takeIndex;
//队尾下标
int putIndex;
//元素个数
int count;
//锁
final ReentrantLock lock;
//出队的条件变量
private final Condition notEmpty;
//入队的条件变量
private final Condition notFull;
...关于它的两个下标变量说明 ：
ArrayBlockingQueue的入队下标和出队下标相当于两个指针，入队操作是在队尾添加元素，出队操作比入队麻烦，出队后需要将剩下的元素向队头移动一个位置，为此优化设计了这两个指针，在分析时把数组看成环状结构，如下图：
出队时将takeIndex下标加1，当takeIndex达到数组末尾时将下标重置为0，入队时将putIndex下标加1，当putIndex达到数组末尾时将下标重置为0。
2、构造方法
构造方法如下：
ArrayBlockingQueue(int capacity)
创建具有给定（固定）容量和默认访问策略的 ArrayBlockingQueue 。
ArrayBlockingQueue(int capacity, boolean fair)
创建一个 ArrayBlockingQueue具有给定（固定）容量和指定访问策略。
ArrayBlockingQueue(int capacity, boolean fair, Collection<? extends E> c)
创建一个 ArrayBlockingQueue具有给定（固定）容量，指定访问策略和最初包含给定集合中的元素，添加在收集迭代器的遍历顺序。第三个构造方法可以设置初始的集合元素。
3、出队与入队
ArrayBlockingQueue的入队与出队方法如下：
入队：
boolean add(E e)
在插入此队列的尾部，如果有可能立即这样做不超过该队列的容量，返回指定的元素 true成功时与抛出 IllegalStateException如果此队列已满。
void put(E e)
在该队列的尾部插入指定的元素，如果队列已满，则等待空间变为可用。
boolean offer(E e)
如果可以在不超过队列容量的情况下立即将其指定的元素插入该队列的尾部，则在成功时 false如果该队列已满，则返回 true 。
boolean offer(E e, long timeout, TimeUnit unit)
在该队列的尾部插入指定的元素，等待指定的等待时间，以使空间在队列已满时变为可用。出队：
E poll()
检索并删除此队列的头，如果此队列为空，则返回 null 。
E poll(long timeout, TimeUnit unit)
检索并删除此队列的头，等待指定的等待时间（如有必要）使元素变为可用。
E take()
检索并删除此队列的头，如有必要，等待元素可用。下边查阅put和take方法的源代码，如下：
public void put(E e) throws InterruptedException {
checkNotNull(e);
//全局锁
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
//队列已满则等待
while (count == items.length)
notFull.await();
//入队
enqueue(e);
} finally {
lock.unlock();
}
}
public E take() throws InterruptedException {
//全局锁
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
//队列为空则等待
while (count == 0)
notEmpty.await();
//出队
return dequeue();
} finally {
lock.unlock();
}
}入队操作：
private void enqueue(E x) {
// assert lock.getHoldCount() == 1;
// assert items[putIndex] == null;
final Object[] items = this.items;
items[putIndex] = x;
//如果put下标达到数组长度则从第0下标开始(循环数组特点)
if (++putIndex == items.length)
putIndex = 0;
count++;//元素个数加1
notEmpty.signal();//唤醒出队的线程
}出队操作：
private E dequeue() {
// assert lock.getHoldCount() == 1;
// assert items[takeIndex] != null;
final Object[] items = this.items;
@SuppressWarnings("unchecked")
E x = (E) items[takeIndex];
//出队元素位置设置为null
items[takeIndex] = null;
if (++takeIndex == items.length)//如果出队下标到数组末尾则将出队下标设置为0
takeIndex = 0;
count--;
if (itrs != null)
itrs.elementDequeued();//更新迭代器中的元素
notFull.signal();//唤醒一个入队的线程
return x;
}2.3 LinkedBlockingQueue
2.3.1 LinkedBlockingQueue介绍
LinkedBlockingQueue与ArrayBlockingQueue不同点：
1、底层的数据结构不同
ArrayBlockingQueue基于数组存储元素，LinkedBlockingQueue基于链表存储元素。
2、队列大小不同
ArrayBlockingQueue构造时必须指定容量，LinkedBlockingQueue构造时不指定容量则默认为Integer.MAX_VALUE，所以它是一个无界队列，也可以定义为一个有界队列。
3、ArrayBlockingQueue使用一把锁，LinkedBlockingQueue使用两把锁，一个出队锁一个入队锁。
4、ArrayBlockingQueue构造时将数组初始化完成，LinkedBlockingQueue则在入队时创建Node元素。
2.3.2 LinkedBlockingQueue源码解析
LinkedBlockingQueue实现了队列、阻塞队列、集合的方法，关于出队和入队的基本操作这里不再演示，下边通过源代码分析它的原理。
LinkedBlockingQueue与LinkedList一样，底层是链表结构，在构造它时如果不指定容量时默认为Integer.MAX_VALUE，所以它是一个无界队列，也可以定义为有界，为了避免不必要的麻烦建议构造时传入队列大小。
如下源代码：
public LinkedBlockingQueue() {
this(Integer.MAX_VALUE);
}
public LinkedBlockingQueue(int capacity) {
if (capacity <= 0) throw new IllegalArgumentException();
this.capacity = capacity;
//默认有一个空结点
last = head = new Node<E>(null);
}下边通过常用方法分析它的原理。
1、put方法
public void put(E e) throws InterruptedException {
//空元素时抛出异常
if (e == null) throw new NullPointerException();
int c = -1;
Node<E> node = new Node<E>(e);
//使用单独的入队锁，在结点数大于等于2时入队与出队并发进行，提高效率
final ReentrantLock putLock = this.putLock;
//维护元素个数
final AtomicInteger count = this.count;
putLock.lockInterruptibly();
try {
while (count.get() == capacity) {
//如果队列满则等待
notFull.await();
}
//入队
enqueue(node);
//计数加1
c = count.getAndIncrement();
//如果队列未满则唤醒其它入队线程提高并发性能
if (c + 1 < capacity)
notFull.signal();
} finally {
putLock.unlock();
}
if (c == 0)//添加前队列为空则有可能有出队线程在等待于是唤醒一个出队线程
signalNotEmpty();
}2、take方法
public E take() throws InterruptedException {
E x;
int c = -1;
final AtomicInteger count = this.count;
//单独使用出队锁，队列元素大于等于2时不影响入队，提高并发性
final ReentrantLock takeLock = this.takeLock;
takeLock.lockInterruptibly();
try {
//队列空则等待
while (count.get() == 0) {
notEmpty.await();
}
//出队
x = dequeue();
//元素计数减1
c = count.getAndDecrement();
//如果队列未空则唤醒其它出队线程提高并发性能
if (c > 1)
notEmpty.signal();
} finally {
takeLock.unlock();
}
if (c == capacity)//添加前队列满则有可能有入队线程在等待于是唤醒一个入队线程
signalNotFull();
return x;
}2.3.3 性能测试
测试代码：
本次测试主要测试多线程提交入队和出队，代码如下：
package com.yjoffer.javase.thread.safe;import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
/**
线程安全集合类测试
@author 预见猿份(www.yjoffer.com)
@version 1.0
/
public class CollectionTest {
/
*ArrayBlockingQueue与LinkedBlockingQueue性能测试- @param queue 队列实例
- @param threadNum 线程个数
- @param number 数据总数
*/
public static void test_array_linked(BlockingQueue<Integer> queue,int threadNum,int number){
ExecutorService threadPool = Executors.newCachedThreadPool();
long start = System.currentTimeMillis();
//多线程提交入队
for (int i = 0; i < threadNum; i++) {
threadPool.submit(()->{
try {
for (int i1 = 0; i1 < number; i1++) {
queue.put(i1);
}
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
CompletionService<Integer> completionService = new ExecutorCompletionService<Integer>(threadPool);
//多线程提交出队
for (int i = 0; i < threadNum; i++) {
completionService.submit(()->{
for (int i1 = 0; i1 < number; i1++) {
queue.take();
}
return 0;
});
}
//等待每个线程的结果
for (int i = 0; i < threadNum; i++) {
try {
completionService.take().get();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
}
//总时长
System.out.print(System.currentTimeMillis() - start);
shutdown(threadPool);
}
public static void main(String[] args) {
System.out.println("ArrayBlockingQueue(ms)\tLinkedBlockingQueue(ms)");
for (int i = 0; i < 10; i++) {
test_array_linked(new ArrayBlockingQueue<>(1000),100,1000000);
System.out.print("\t\t\t\t\t\t");
test_array_linked(new LinkedBlockingQueue<>(1000),100,1000000);
System.out.println();
}
}</code></pre>测试用例1：一百万个数据，100个线程，队列容量1000
测试结果如下：
ArrayBlockingQueue(ms) LinkedBlockingQueue(ms)
22519 12950
17336 14954
18028 14502
19788 14834
15751 14620
32968 14617
15350 15051该测试用例下LinkedBlockingQueue的速度略高于ArrayBlockingQueue。
测试用例2：一百万个数据，50个线程，队列容量1000
ArrayBlockingQueue(ms) LinkedBlockingQueue(ms)
8354 7007
7128 7352
7067 6878
7947 7193
8028 7193
9595 6978
8907 6750
8130 6958当线程数减少原来的一半时下LinkedBlockingQueue的速度与ArrayBlockingQueue接近。
测试用例3：一百万个数据，50个线程，队列容量10000
ArrayBlockingQueue(ms) LinkedBlockingQueue(ms)
3371 7415
3877 7236
3576 6808
3637 7180
3626 7150
3722 6930当线程容量增加为原来的10倍时ArrayBlockingQueue的速度高于LinkedBlockingQueue。
结论：
当队列的数据规模不大，并发线程不多时建议使用ArrayBlockingQueue。
当队列的数据规模非常大，并发线程多时建议使用LinkedBlockingQueue。
2.4 LinkedBlockingDeque
2.4.1 LinkedBlockingDeque介绍
LinkedBlockingDeque和LinkedBlockingQueue的底层都是链表结构，LinkedBlockingQueue只能在一端入队，在另一端出队，这是单向链表，LinkedBlockingDeque是双向阻塞队列，两边都可以出入队，这是双向链表。LinkedBlockingDeque实现了Deque接口。
查看LinkedBlockingDeque的API发现出队和入队方法都是两个：XXXFirst、XXXLast，以First结尾的方法表示操作第一个元素，以Last结尾的方法表示操作结尾的元素，比如：putFirst在队首入队，putLast在队尾入队。
LinkedBlockingDeque并不像LinkedBlockingQueue有两把锁，LinkedBlockingDeque只使用一个ReentrantLock锁，同时只有一个线程可以在队列一端出队或入队。
下边通过源代码看下它的工作原理：
public void putFirst(E e) throws InterruptedException {
//结点不允许为空
if (e == null) throw new NullPointerException();
Node<E> node = new Node<E>(e);
final ReentrantLock lock = this.lock;
lock.lock();
try {
//入队失败则等待，队列有空位则继续
while (!linkFirst(node))
notFull.await();
} finally {
lock.unlock();
}
}linkFist将元素插在队头，源代码如下：
private boolean linkFirst(Node<E> node) {
// assert lock.isHeldByCurrentThread();
//队列满则终止
if (count >= capacity)
return false;
Node<E> f = first;
//原来队头结点变为下一个
node.next = f;
//first指向新结点
first = node;
if (last == null)
last = node;
else
f.prev = node;
++count;
//通知消费者线程
notEmpty.signal();
return true;
}2.4.2 LinkedBlockingDeque应用
双端队列凭借两端都可入队或出队的特点在生产中可应用在“工作窃取模式”，"工作窃取"模式是指一个线程窃取另一个线程的工作，比如在生产者与消费者应用中，每个消费者对应一个工作队列，当消费者自己工作队列中没有任务时它会去其它线程的工作队列 去消费，为了提高并发它会去队尾消费一个任务，这里就是应用了双端队列 的特点，如下图：
消费者1与消费者2两个线程分别去自己的工作消费任务，同时消费者2线程会去消费者1的工作队列进行工作窃取。
测试代码如下：
package com.yjoffer.javase.thread.safe;
import com.yjoffer.javase.config.Logger;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
/**
* 线程安全集合类测试
* @author 预见猿份(www.yjoffer.com)
* @version 1.0
**/
public class BlockingQueueTest {
private static Logger logger = Logger.getLogger(BlockingQueueTest.class);
//测试LinkedBlockingDeque，工作窃取
public static void test_LinkedBlockingDeque(){
ExecutorService threadPool = Executors.newCachedThreadPool();
//消费队列1
LinkedBlockingDeque<Integer> linkedBlockingDeque1 = new LinkedBlockingDeque<>();
//消费队列2
LinkedBlockingDeque<Integer> linkedBlockingDeque2 = new LinkedBlockingDeque<>();
//生产一批数据
threadPool.execute(()->{
//队列一10个数
for (int i = 0; i < 10; i++) {
//put调用putLast
try {
linkedBlockingDeque1.put(i);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
//队列二20个数
for (int i = 10; i < 30; i++) {
try {
linkedBlockingDeque2.put(i);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
//消费者1
threadPool.execute(()->{
while (true){
if (!linkedBlockingDeque1.isEmpty()) {
// 从队头消费自己队列的数据
try {
logger.info(String.valueOf(linkedBlockingDeque1.takeFirst()));
} catch (InterruptedException e) {
e.printStackTrace();
}
} else if (!linkedBlockingDeque2.isEmpty()) {
// 工作窃取其它队列，从队尾开始
try {
logger.info("窃取"+linkedBlockingDeque2.takeLast());
} catch (InterruptedException e) {
e.printStackTrace();
}
} else {
break;
}
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
//消费者2
threadPool.execute(()->{
while (true){
if (!linkedBlockingDeque2.isEmpty()) {
// 从队头消费自己队列的数据
try {
logger.info(String.valueOf(linkedBlockingDeque2.takeFirst()));
} catch (InterruptedException e) {
e.printStackTrace();
}
} else if (!linkedBlockingDeque1.isEmpty()) {
// 工作窃取其它队列，从队尾开始
try {
logger.info("窃取"+linkedBlockingDeque1.takeLast());
} catch (InterruptedException e) {
e.printStackTrace();
}
} else {
break;
}
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
shutdown(threadPool);
}
public static void main(String[] args) {
test_LinkedBlockingDeque();
}
}运行程序，输出 如下：
19:53:33.325 INFO com.yjoffer.javase.thread.safe.CollectionTest[pool-1-thread-2] - 0
19:53:33.326 INFO com.yjoffer.javase.thread.safe.CollectionTest[pool-1-thread-1] - 10
19:53:34.348 INFO com.yjoffer.javase.thread.safe.CollectionTest[pool-1-thread-2] - 1
19:53:34.348 INFO com.yjoffer.javase.thread.safe.CollectionTest[pool-1-thread-1] - 11
19:53:35.348 INFO com.yjoffer.javase.thread.safe.CollectionTest[pool-1-thread-1] - 12
19:53:35.348 INFO com.yjoffer.javase.thread.safe.CollectionTest[pool-1-thread-2] - 2
19:53:36.348 INFO com.yjoffer.javase.thread.safe.CollectionTest[pool-1-thread-2] - 3
19:53:36.348 INFO com.yjoffer.javase.thread.safe.CollectionTest[pool-1-thread-1] - 13
19:53:37.348 INFO com.yjoffer.javase.thread.safe.CollectionTest[pool-1-thread-2] - 4
19:53:37.349 INFO com.yjoffer.javase.thread.safe.CollectionTest[pool-1-thread-1] - 14
19:53:38.348 INFO com.yjoffer.javase.thread.safe.CollectionTest[pool-1-thread-2] - 5
19:53:38.349 INFO com.yjoffer.javase.thread.safe.CollectionTest[pool-1-thread-1] - 15
19:53:39.348 INFO com.yjoffer.javase.thread.safe.CollectionTest[pool-1-thread-2] - 6
19:53:39.349 INFO com.yjoffer.javase.thread.safe.CollectionTest[pool-1-thread-1] - 16
19:53:40.348 INFO com.yjoffer.javase.thread.safe.CollectionTest[pool-1-thread-2] - 7
19:53:40.349 INFO com.yjoffer.javase.thread.safe.CollectionTest[pool-1-thread-1] - 17
19:53:41.348 INFO com.yjoffer.javase.thread.safe.CollectionTest[pool-1-thread-2] - 8
19:53:41.349 INFO com.yjoffer.javase.thread.safe.CollectionTest[pool-1-thread-1] - 18
19:53:42.348 INFO com.yjoffer.javase.thread.safe.CollectionTest[pool-1-thread-2] - 9
19:53:42.349 INFO com.yjoffer.javase.thread.safe.CollectionTest[pool-1-thread-1] - 19
19:53:43.348 INFO com.yjoffer.javase.thread.safe.CollectionTest[pool-1-thread-2] - 窃取29
19:53:43.349 INFO com.yjoffer.javase.thread.safe.CollectionTest[pool-1-thread-1] - 20
19:53:44.349 INFO com.yjoffer.javase.thread.safe.CollectionTest[pool-1-thread-1] - 21
19:53:44.349 INFO com.yjoffer.javase.thread.safe.CollectionTest[pool-1-thread-2] - 窃取28
19:53:45.349 INFO com.yjoffer.javase.thread.safe.CollectionTest[pool-1-thread-2] - 窃取27
19:53:45.349 INFO com.yjoffer.javase.thread.safe.CollectionTest[pool-1-thread-1] - 22
19:53:46.349 INFO com.yjoffer.javase.thread.safe.CollectionTest[pool-1-thread-1] - 23
19:53:46.349 INFO com.yjoffer.javase.thread.safe.CollectionTest[pool-1-thread-2] - 窃取26
19:53:47.349 INFO com.yjoffer.javase.thread.safe.CollectionTest[pool-1-thread-1] - 24
19:53:47.350 INFO com.yjoffer.javase.thread.safe.CollectionTest[pool-1-thread-2] - 窃取25从输出结果可以看出pool-1-thread-2窃取了5个数。
2.5 DelayQueue
2.5.1 DelayQueue介绍
DelayQueue是延迟队列，队列中的每一个元素都有一个延迟时间，根据延迟时间可以判断元素是否到期，如果到期可以出队，如下图：
delayQueue队列中有三个元素: E1、E2、E3，E1的延迟时间为2秒，E2的延迟时间为4秒，E3的延迟时间为6秒，从元素入队开始算，经过延迟时间后出队，假设E1入队的时间点为0点整，延迟2秒就是0点0分2秒，当到达0点0分2秒时E1出队，所以上图中E1先出队，而后是E2、E3。
延迟队列中的元素在经过延迟时间后出队，利用它延迟的特性在项目场景中应用广泛：
1、12306网站中未支付订单保留45分钟，也就是说未支付订单延迟45分钟后自动取消。
2、网上门诊挂号预约成功，60秒后发送提示短信。
3、物流变更，60秒后通知用户。
2.5.2 Delayed接口
DelayQueue延迟队列中放的元素必须具有延迟时间属性，所以延迟队列的元素必须实现Delayed接口，如下：
public interface Delayed extends Comparable<Delayed> {
long getDelay(TimeUnit unit);
}getDelay方法：返回该任务的剩余延迟时间，形参unit表示按给定时间单位去计算。用到期时间点减去当前时间点就是getDelay的值，如果为0或负数则说明该任务延迟时间到期可以执行，否则还需要继续等待。
compareTo方法：Delayed接口继承了Comparable接口，该接口中有一个compareTo方法，它的作用是对延迟队列中的任务进行排序，方法如下：
public interface Comparable<T> {
public int compareTo(T o);
}Comparable接口在“集合”章节讲解过，Comparable接口的compareTo方法会在进行元素比较时调用。
compareTo方法使用技巧：
当this.o.XXX >o.XXX则返回大于0的数，当this.o.XXX<o.XXX则返回小于0的数，两者相等返回0。
2.5.3 DelayQueue测试
1、定义延迟队列任务类
package com.yjoffer.javase.thread.basic;
import com.yjoffer.javase.config.Logger;
import java.util.concurrent.*;
/**
* BlockingQueue测试
* @author 预见猿份(www.yjoffer.com)
*/
public class BlockingQueueTest {
private static Logger logger = Logger.getLogger(BlockingQueueTest.class);
static class DelayedTask1 implements Delayed{
private String content;//任务内容
private long delay; //延迟时间,单位秒
private long expire; //到期时间，毫秒
/**
*
* @param delay 延迟时间(秒)
* @param content 任务内容
*/
public DelayedTask1(long delay,String content) {
this.delay = delay;
this.content = content;
expire = System.currentTimeMillis() + delay*1000;
}
public String getContent() {
return content;
}
/**
* 比较方法 ，delayQueue使用此方法进行排序
* 以x.compareTo(Delayed o)为例，当x>o 则返回大于0的数；当x<o则返回小于0的数
* @param o
* @return
*/
@Override
public int compareTo(Delayed o) {
//升序，按延迟时间由小到大排序，延迟时间最小的先出队
return (int) (this.getDelay(TimeUnit.MILLISECONDS) -o.getDelay(TimeUnit.MILLISECONDS));
}
//返回给定时间单位的剩余延时时间，<=0表示到期可以执行，>0表示未到期
@Override
public long getDelay(TimeUnit unit) {
//将this.expire - System.currentTimeMillis()的毫秒数转为指定单位的数量
return unit.convert(this.expire - System.currentTimeMillis() , TimeUnit.MILLISECONDS);
}
}2、测试
//测试DelayQueue
public static void test_delayQueue() {
DelayQueue<DelayedTask1> delayQueue = new DelayQueue<>();
delayQueue.put(new DelayedTask2(2,"02任务"));
delayQueue.put(new DelayedTask2(3,"03任务"));
delayQueue.put(new DelayedTask2(4,"04任务"));
delayQueue.put(new DelayedTask2(5,"05任务"));
delayQueue.put(new DelayedTask2(1,"01任务"));
ExecutorService threadPool = Executors.newCachedThreadPool();
for (int i = 0; i < 5; i++) {
threadPool.execute(()->{
try {
DelayedTask2 task = delayQueue.take();
logger.debug("消费任务"+task.getContent());
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
}
public static void main(String[] args) {
test_delayQueue();
}
}地程序，输出如下 ：
18:42:53.65 FINE com.yjoffer.javase.thread2.basic.BlockingQueueTest[pool-1-thread-1] - 消费任务01任务
18:42:54.64 FINE com.yjoffer.javase.thread2.basic.BlockingQueueTest[pool-1-thread-2] - 消费任务02任务
18:42:55.64 FINE com.yjoffer.javase.thread2.basic.BlockingQueueTest[pool-1-thread-5] - 消费任务03任务
18:42:56.65 FINE com.yjoffer.javase.thread2.basic.BlockingQueueTest[pool-1-thread-3] - 消费任务04任务
18:42:57.65 FINE com.yjoffer.javase.thread2.basic.BlockingQueueTest[pool-1-thread-4] - 消费任务05任务从输出可以看出01任务在07秒执行，2秒后02任务执行，而后是03任务、04任务。
2.5.4 DelayQueue源码分析
下边通过查看DelayQueue的源代码分析延迟队列的原理。
入队：
public boolean add(E e) {
return offer(e);
}
public void put(E e) {
offer(e);
}
//其它方法调用offer方法入队
public boolean offer(E e) {
final ReentrantLock lock = this.lock;
lock.lock();
try {
//入队，底层是优先级队列，无界队列
q.offer(e);
//如果添加的元素为队首元素此时唤醒消费者
if (q.peek() == e) {
leader = null;
available.signal();
}
return true;
} finally {
lock.unlock();
}
}出队：
区分阻塞与非阻塞。
非阻塞方法：
public E poll() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
E first = q.peek();
//如果没有检索到队首元素或者元素还没有到期返回null
if (first == null || first.getDelay(NANOSECONDS) > 0)
return null;
else
//检索到队首元素且元素到期则出队
return q.poll();
} finally {
lock.unlock();
}
}阻塞方法：
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
//可打断锁
lock.lockInterruptibly();
try {
for (;;) {
E first = q.peek();
//如果队首元素为null则线程等待
if (first == null)
available.await();
else {
//取出队头元素的剩余延迟时间
long delay = first.getDelay(NANOSECONDS);
//如果剩余延迟时间小于等于0说明到期则立即出队
if (delay <= 0)
return q.poll();
//first不用时设置为null，有利于GC，可能队头元素被其它线程出队
first = null; // don't retain ref while waiting
if (leader != null)//如果当前有线程在等待出队则当前线程也进入等待队列
available.await();
else {//否则当前线程开始等待
Thread thisThread = Thread.currentThread();
leader = thisThread;
try {
//等待指定的延期时长，awaitNanos方法使当前线程等待直到发出信号或中断，或指定的等待时间过去。
available.awaitNanos(delay);
} finally {
//将leader设置为空让其它线程取元素
if (leader == thisThread)
leader = null;
}
}
}
}
} finally {
//如果leader为空或者队头有元素则唤醒其它等待取元素的线程
if (leader == null && q.peek() != null)
available.signal();
lock.unlock();
}
}2.6 SynchronousQueue
2.6.1 SynchronousQueue介绍
SynchronousQueue是一个特殊的阻塞队列，它的特殊在于它没有使用队列存储数据，前边学习的ArrayBlockingQueue使用数组存储数据、LinkedBlockingQueue使用链表存储数据，而SynchronousQueue是生产者与消费者同步交互的媒介并不存储它们之间交互的数据。
为什么说SynchronousQueue是生产者与消费者同步交互的媒介呢？正如它的名字一样，它是一个同步队列，下边通过生产者与消费者通过SynchronousQueue交互的例子说明：
1、生产者调用put方法向同步队列写入一条数据，此时阻塞
2、消费者调用take方法从同步队列读出一条数据，此时put方法解除阻塞。
所以：put和take方法必须成对出现，先调用put方法则put方法阻塞，必须由take方法解除阻塞；如果先调用take方法则take方法阻塞必须由put方法解除阻塞。
2.6.2 SynchronousQueue测试
下边测试SynchronousQueue的特点。
下边的程序分别测评先运行put方法以及先运行take方法的情况。
package com.yjoffer.javase.thread.basic;
import com.yjoffer.javase.config.Logger;
import java.util.Collection;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.*;
/**
* BlockingQueue测试
* @author 预见猿份(www.yjoffer.com)
*/
public class BlockingQueueTest {
private static Logger logger = Logger.getLogger(BlockingQueueTest.class);
//测试SynchronousQueue
public static void test_SynchronousQueue(){
SynchronousQueue<String> queue = new SynchronousQueue<>();
ExecutorService threadPool = Executors.newCachedThreadPool();
threadPool.execute(()->{
try {
queue.put("hello");
} catch (InterruptedException e) {
e.printStackTrace();
}
});
threadPool.execute(()->{
try {
TimeUnit.SECONDS.sleep(2);
String take = queue.take();
logger.debug(take);
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
public static void main(String[] args) {
test_SynchronousQueue();
}
}1、首先测试先运行put方法
在queue.take()代码前加TimeUnit.SECONDS.sleep(2);，加上如下断点：
断点选择"Thread":
运行程序，首先在put方法处阻塞，此时执行queue.take()后put方法阻塞解除，并成功获了put方法写入的元素。
2、测试先运行take方法
在queue.put()代码前加TimeUnit.SECONDS.sleep(2);，断点 位置不变。
运行程序，首先在take()方法处阻塞，此时执行queue.put()后take()方法阻塞解除，并成功获了put方法写入的元素。
2.6.3 SynchronousQueue应用
SynchronousQueue的put方法和take方法是在同步发生，当两个线程想要同步传递数据时可以使用SynchronousQueue。如下图：
生产者线程调用put方法向消费者线程传递数据，消费者线程调用take方法获取传递的数据。
有时候这种传递是双向，即生产者请求消费者、消费者响应生产者，如下图：
下边测试双向数据传递的情况：
线程1向线程2发送“预见猿份”，线程2收到请求并响应“www.yjoffer.com”，代码如下：
//测试SynchronousQueue
public static void test_SynchronousQueue2(){
SynchronousQueue<String> queue = new SynchronousQueue<>();
ExecutorService threadPool = Executors.newCachedThreadPool();
threadPool.execute(()->{
try {
String requestbody = "预见猿份";
queue.put(requestbody);
logger.debug("request.."+requestbody);
String take = queue.take();
logger.debug("receive response.."+take);
} catch (InterruptedException e) {
e.printStackTrace();
}
});
threadPool.execute(()->{
try {
String take = queue.take();
logger.debug("receive request.."+take);
String responsebody = "www.yjoffer.com";
queue.put(responsebody);
logger.debug("response.."+responsebody);
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}运行程序：
15:48:42.643 FINE com.yjoffer.javase.thread.basic.BlockingQueueTest[pool-1-thread-1] - request..预见猿份
15:48:42.643 FINE com.yjoffer.javase.thread.basic.BlockingQueueTest[pool-1-thread-2] - receive request..预见猿份
15:48:42.666 FINE com.yjoffer.javase.thread.basic.BlockingQueueTest[pool-1-thread-2] - response..www.yjoffer.com
15:48:42.666 FINE com.yjoffer.javase.thread.basic.BlockingQueueTest[pool-1-thread-1] - receive response..www.yjoffer.com2.7 CopyOnWriteArrayList
2.7.1 CopyOnWriteArrayList介绍
在jdk中有一类以CopyOnWrite开头的集合，字面意思为“写入时复制”，它的思想是：当一个线程去修改集合中的内容时将集合内容先拷贝一份，在拷贝中去修改，修改后再将原集合的引入指向拷贝的集合。
jdk中提供了两个CopyOnWirte容器：CopyOnWriteArrayList、CopyOnWriteArraySet。
CopyOnWriteArraySet的底层使用了CopyOnWriteArrayList，CopyOnWriteArraySet保证了元素的唯一性。
下面读取CopyOnWriteArrayList的源代码(jdk1.8)理解它的原理：
//向集合添加一个元素
public boolean add(E e) {
final ReentrantLock lock = this.lock;
lock.lock();
try {
Object[] elements = getArray();
int len = elements.length;
//拷贝一个新数组
Object[] newElements = Arrays.copyOf(elements, len + 1);
//将新元素添加到拷贝数组中
newElements[len] = e;
//将原数组引用指向新数组
setArray(newElements);
return true;
} finally {
lock.unlock();
}从源代码可以看出写方法使用了ReentrantLock锁。
下边是读方法，读方法中没有使用锁。
public E get(int index) {
return get(getArray(), index);
}
private E get(Object[] a, int index) {
return (E) a[index];
}CopyOnWrite容器的应用场景：
CopyOnWrite可实现读读并发、读写并发，由于每次写都会拷贝一份所以适合读多写少的场景，比如：系统配置信息、商品目录信息、商家地址信息等，这些信息变化较少，无锁读取速度很快。
2.7.2 CopyOnWriteArrayList优势
CopyOnWirte集合的好处如下：
1、并发读
由于修改操作在拷贝的集合中进行，所以从原集合读取元素不用加锁，读和写可并发进行。
2、读写分离
很多容器是不支持读写并发的，CopyOnWrite则可以实现读写并发，因为读线程从原集合中读数据，写线程向拷贝集合中写数据，读写是不同的集合，读写是分离的。
2.7.3 CopyOnWriteArrayList弱一致
CopyOnWriteArrayList的缺点如下：
1、内存占用多
由于在写入时内存中同时维护两份数据，给内存造成一定的压力，写完后原数据无效给垃圾收集造成一定的压力。
2、弱一致性
弱一致性是指读与写之间会存在数据不一致的现象，但整体上是一致的。强一致性是指数据始终是一致的。
比如下边的代码体现的弱一致性：
package com.yjoffer.javase.thread.basic;
import com.yjoffer.javase.config.Logger;import java.util.Iterator;
import java.util.concurrent.*;
/**
BlockingQueue测试
@author 预见猿份(www.yjoffer.com)
*/
public class BlockingQueueTest {
private static Logger logger = Logger.getLogger(BlockingQueueTest.class);
//测试CopyOnWrite的弱一致性
public static void test_copyonwrite(){//创建copyonwrite集合 CopyOnWriteArrayList arrayList = new CopyOnWriteArrayList(); arrayList.add("yjoffer"); arrayList.add(".");
arrayList.add("com");
//迭代器
Iterator iterator = arrayList.iterator();
//新线程修改数据
Thread thread = new Thread(() -> {
arrayList.add(0,"www");
});
thread.start();
try {
thread.join();
} catch (InterruptedException e) {
e.printStackTrace();
}
//使用迭代器遍历
iterator.forEachRemaining(item->{
System.out.println(item);
});
}
public static void main(String[] args) {
test_copyonwrite();
}}
运行程序，输出 ：
yjoffer
.
com新线程添加的"www"并没有在迭代器中立即生效，这就是弱一致性的体现。
要想尽可能的遍历最新的数据建议遍历前获取迭代器，代码如下：
//使用迭代器遍历
arrayList.iterator().forEachRemaining(item->{
System.out.println(item);
});