【并发编程】5.原子类与并发容器

一、原子类

1.什么是原子类

Java的java.util.concurrent包除了提供底层锁、并发集合外，还提供了一组原子操作的封装类，它们位于java.util.concurrent.atomic包。

Atomic类是通过无锁（lock-free）的方式实现的线程安全（thread-safe）访问

2.原子类是基于什么实现的

原子类的实现是基于CPU本身提供的 CAS指令,是一种无锁的实现；

CAS Compare and Swap 比较并交换，即在更新之前会校验是否于期待的值相等

执行函数：CAS(V,E,N)

V表示要更新的变量；E表示预期值；N表示新值

CAS 容易出现ABA问题

容易出现在原子变量值不确定容易反复的情况

假设 count 原本是 A，

线程1检查数值 期望值为 A，未执行更新操作 ，

此时 线程2 更新值为B，

线程3更新值 为 A，

然后 线程1执行更新操作 ，期望值与实际值相同执行操作。

解决方案：类似乐观锁，加上版本号。

3.常用的原子类

1. 原子化的基本数据类型

相关实现有 AtomicBoolean、AtomicInteger 和 AtomicLong

2. 原子化的对象引用类型

相关实现有 AtomicReference、AtomicStampedReference 和 AtomicMarkableReference，利用 它们可以实现对象引用的原子化更新。

ABA 问题

AtomicStampedReference 和 AtomicMarkableReference 这两个原子类可以解决 ABA 问题

3. 原子化数组

相关实现有 AtomicIntegerArray、AtomicLongArray 和 AtomicReferenceArray，利用这些原子 类，我们可以原子化地更新数组里面的每一个元素。这些类提供的方法和原子化的基本数据类型 的区别仅仅是：每个方法多了一个数组的索引参数，所以这里也不再赘述了。

4. 原子化对象属性更新器

相关实现有 AtomicIntegerFieldUpdater、AtomicLongFieldUpdater 和 AtomicReferenceFieldUpdater，利用它们可以原子化地更新对象的属性，这三个方法都是利用反 射机制实现的。

5. 原子化的累加器

DoubleAccumulator、DoubleAdder、LongAccumulator 和 LongAdder，这四个类仅仅用来执行 累加操作，相比原子化的基本数据类型，速度更快，但是不支持 compareAndSet() 方法。如果你 仅仅需要累加操作，使用原子化的累加器性能会更好

4.原子类的用途

适用于计数器，累加器等场景（感觉redis更好用..）

二、并发容器

1.同步容器

java 1.5之前在java.util包中提供了Vector和HashTable两个同步容器

是对所有的方法都加上了同步关键字synchronized，确保一个实例只有一个线程能操作。

这样所有的操作就都是串行的，性能较差。

2.常用的并发容器与实现原理

2.1 CopyOnWriteArrayList 是唯一的并发List 读操作完全无锁

/** The lock protecting all mutators */
final transient ReentrantLock lock = new ReentrantLock(); //可重入锁
/** The array, accessed only via getArray/setArray. */
private transient volatile Object[] array; //内部维护的对象数组
public E get(int index) {
return get(getArray(), index);
}
@SuppressWarnings("unchecked")
private E get(Object[] a, int index) {
return (E) a[index];
}


//进行写操作时会把当前数组copy一份，进行写操作后将新的数组赋值给 array
public E set(int index, E element) {
final ReentrantLock lock = this.lock;
lock.lock();
try {
Object[] elements = getArray();
E oldValue = get(elements, index);
if (oldValue != element) {
int len = elements.length;
Object[] newElements = Arrays.copyOf(elements, len);
newElements[index] = element;
setArray(newElements);
} else {
// Not quite a no-op; ensures volatile write semantics
setArray(elements);
}
return oldValue;
} finally {
lock.unlock();
}
}
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();
}
}


CopyOnWriteArrayList 迭代器是只读的，不支持增删改。因为迭代器遍历的仅仅是一个快照，而对快照进行增删改是没 有意义的。

2.2.ConcurrentHashMap(key 无序)和 ConcurrentSkipListMap(key 有序)

ConcurrentHashMap与ConcurrentSkipListMap中key value 都不能为空，否则会空指针

ConcurrentHashMap容器相较于CopyOnWrite容器在并发加锁粒度上有了更大一步的优化，它通过修改对单个hash桶元素加锁的达到了更细粒度的并发控制。

• 在发生hash冲突时仅仅只锁住当前需要添加节点的头元素即可，可能是链表头节点或者红黑树的根节点，其他桶节点都不需要加锁，大大减小了锁粒度。


• ConcurrentHashMap容器是通过CAS + synchronized一起来实现并发控制的。

/** Implementation for put and putIfAbsent */
final V putVal(K key, V value, boolean onlyIfAbsent) {
if (key == null || value == null) throw new NullPointerException();
int hash = spread(key.hashCode()); //计算key的hash值
int binCount = 0;
//循环插入元素，避免并发插入失败
for (Node[] tab = table;;) {
Node f; int n, i, fh;//f是hash桶的头结点
if (tab == null || (n = tab.length) == 0)
tab = initTable();
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
//如果当前hash桶无元素，通过CAS操作插入新节点
if (casTabAt(tab, i, null,
new Node(hash, key, value, null)))
break; // no lock when adding to empty bin
}
//不断循环计算table（散列表）的每个桶位（slot）的散列值i ,直到找到tab[i] 为空的桶位，casTabAt将put（增加）的节点Node 放到空仓（empty bin）中，如果在put 的过程中，别的线程更改了tab[i],导致tab[i] 不为空，那么casTabAt返回false，继续循环找tab[i]== null的桶位。
//如果当前桶正在扩容，则协助扩容
else if ((fh = f.hash) == MOVED)
tab = helpTransfer(tab, f);
else {
V oldVal = null;
//hash冲突时锁住当前需要添加节点的头元素，可能是链表头节点或者红黑树的根节点
synchronized (f) {
if (tabAt(tab, i) == f) {
if (fh >= 0) {
binCount = 1;
for (Node e = f;; ++binCount) {
K ek;
if (e.hash == hash &&
((ek = e.key) == key ||
(ek != null && key.equals(ek)))) {
oldVal = e.val;
if (!onlyIfAbsent)
e.val = value;
break;
}
Node pred = e;
if ((e = e.next) == null) {
pred.next = new Node(hash, key,
value, null);
break;
}
}
}
else if (f instanceof TreeBin) {
Node p;
binCount = 2;
if ((p = ((TreeBin)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;
}


2.3 Set

Set 接口的两个实现是

CopyOnWriteArraySet 参考CopyOnWriteArrayList

ConcurrentSkipListSe 参考ConcurrentSkipListMap

2.4 Queue

阻塞与非阻塞，所谓阻塞指的是当队列已满时，入队操作阻塞；当队列已空时，出队操作阻塞。

单端与双端，单端指的是只能队尾入队，队首出队；而双端指的是队首队尾皆可入 队出队。

阻塞队列都用 Blocking 关键字标识，单端队列使用 Queue 标识，双端队 列使用 Deque 标识。

1.单端阻塞队列：

• ArrayBlockingQueue、


• LinkedBlockingQueue、


• SynchronousQueue、


• LinkedTransferQueue、


• PriorityBlockingQueue、


• DelayQueue。

内部一般会持有一个队列，这个队列可以是数组（其实现是 ArrayBlockingQueue）也可以是链表（其实 现是 LinkedBlockingQueue）；甚至还可以不持有队列（其实现是 SynchronousQueue），

LinkedTransferQueue融合了LinkedBlockingQueue、 SynchronousQueue功能 性能更好

PriorityBlockingQueue 支持按优先级出队

DelayQueue 支持延时出队

2.双端阻塞队列：

其实现是 LinkedBlockingDeque。

3.单端非阻塞队列：

其实现是 ConcurrentLinkedQueue。

4.双端非阻塞队列：

其实现是 ConcurrentLinkedDeque。

阻塞队列相关API方法的区别