一、CopyOnWrite(COW算法的容器)
最终一致性、写分离思想。
用Volatile修饰,每次直接从内存地址中读取,读取时不加锁。
写时用显式锁整个容器(防止其它写线程),然后拷贝一份副本,对副本操作,读线程访问原容器数据。
内存开销大,实时数据一致性不高。适用于读远大于写操作且数据量不大的场景。
1、CopyOnWriteArrayList
底层是通过数组来实现的
1.构造方法
/**
* 无参构造方法
*/
public CopyOnWriteArrayList() {
setArray(new Object[0]);
}
final void setArray(Object[] a) {
array = a;
}
final Object[] getArray() {
return array;
}
/**
* 传入集合构造方法
*/
public CopyOnWriteArrayList(Collection<? extends E> c) {
Object[] elements;
if (c.getClass() == CopyOnWriteArrayList.class){
elements = ((CopyOnWriteArrayList<?>)c).getArray();
} else {
elements = c.toArray();
if (c.getClass() != ArrayList.class)
elements = Arrays.copyOf(elements, elements.length, Object[].class);
}
setArray(elements);
}
/**
* 传入数组构造方法
*/
public CopyOnWriteArrayList(E[] toCopyIn) {
setArray(Arrays.copyOf(toCopyIn, toCopyIn.length, Object[].class));
}
2.add、set 方法
/**
* 新增元素
*/
public boolean add(E e) {
// 获取当前实例的可重入锁
final ReentrantLock lock = this.lock;
// 阻塞式加锁
lock.lock();
try {
// 获取当前集合的底层数组
Object[] elements = getArray();
int len = elements.length;
// 将当前数组的全部数据拷贝到新数组中,新数组长度为当前长度+1
Object[] newElements = Arrays.copyOf(elements, len + 1);
// 设置新增数据到数组中
newElements[len] = e;
// 将当前实例的底层数组指向新数组
setArray(newElements);
return true;
} finally {
lock.unlock();
}
}
/**
* 修改元素
*/
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();
}
}
2.remove方法
public E remove(int index) {
// 获取锁
final ReentrantLock lock = this.lock;
lock.lock();
try {
Object[] elements = getArray();
int len = elements.length;
// 从数组中获取需要删除的数据
E oldValue = get(elements, index);
// 计算需要移动元素的索引值
int numMoved = len - index - 1;
if (numMoved == 0)
// 需要移动元素的索引值为0表示待删除数据为最后一个元素,拷贝至新数组即可
setArray(Arrays.copyOf(elements, len - 1));
else {
// 需要移动的索引值大于0则需要分段拷贝至新数组
Object[] newElements = new Object[len - 1];
System.arraycopy(elements, 0, newElements, 0, index);
System.arraycopy(elements, index + 1, newElements, index, numMoved);
setArray(newElements);
}
return oldValue;
} finally {
lock.unlock();
}
}
2、CopyOnWriteArraySet
基于 CopyOnWriteArrayList 实现,
不同是在 add 时调用的是 CopyOnWriteArrayList 的addIfAbsent方法,遍历当前Object数组,
如Object数组中已有了当前元素,则直接返回,如果没有则放入Object数组的尾部,并返回。
二、CurrentMap(并发映射)
1、ConcurrentHashMap(代替HashMap)
1.8前:数组 + 链表 + 分段锁(默认16个segement) +lock锁 + unsafe类。get不加锁用volatile + CAS,put加锁控制在小范围内。
1.8后:Node数组+链表+红黑树结构,链表长度超过8时转为红黑树。在锁的实现上,采用CAS + synchronized,复杂逻辑的流程对Node节点用synchronize进行同步。
1.put 方法
没有用synchronized修饰,用 volatille 和 CAS 算法。
1.根据 key 计算出 hash 值;
2.判断是否需要进行初始化;
3.定位到 Node,拿到首节点 f,判断首节点 f:
4.如果为 null ,则通过 CAS 的方式尝试添加;
5.如果为 f.hash = MOVED = -1 ,说明其他线程在扩容,参与一起扩容;
6.如果都不满足 ,synchronized 锁住 f 节点,判断是链表还是红黑树,遍历插入;
7.当在链表长度达到 8 的时候,数组扩容或者将链表转换为红黑树。
public V put(K key, V value) {
return putVal(key, value, false);
}
final V putVal(K key, V value, boolean onlyIfAbsent) {
if (key == null || value == null) throw new NullPointerException();
//获取hash值
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)
tab = initTable();
//如果槽位中为空的
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
//以cas方式进行替换,替换成功就中断循环,替换失败则进行下一次循环
if (casTabAt(tab, i, null,new Node<K,V>(hash, key, value, null)))
break; // no lock when adding to empty bin
}
//一种特殊的节点(forwarding 节点,迁移节点,只在迁移过程中存在)的处理方式
else if ((fh = f.hash) == MOVED)
//帮助进行扩容
tab = helpTransfer(tab, f);
//如果槽位不为空,并且不是(forwarding节点)
else {
V oldVal = null;
//将整个槽位锁住
synchronized (f) {
//double check,如果槽位里面的数据发生变更则重新走流程
if (tabAt(tab, i) == f) {
//如果是链表
if (fh >= 0) {
binCount = 1;
for (Node<K,V> 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<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;
if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
value)) != null) {
oldVal = p.val;
if (!onlyIfAbsent)
p.val = value;
}
}
}
}
//如果一个槽位中的数量大于1(只有大于1的才会有binCount)
if (binCount != 0) {
//如果槽位中的成员数量大于等于8,则变更为
if (binCount >= TREEIFY_THRESHOLD)
//进行转换成红黑树处理
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
}
}
addCount(1L, binCount);
return null;
}
/**
* 获取槽位
*/
static final int spread(int h) {
//h为key值得hash值,将高16位也参与运算,然后与int最大值进行&运算(效果为将值变为正数,其他位置不变)
//HASH_BITS为int最大值,最高位为0
//HashMap中没有处理为正数的步骤,这里负数有其它含义,查看节点类型
return (h ^ (h >>> 16)) & HASH_BITS;
}
/**
* 初始化容器
*/
private final Node<K,V>[] initTable() {
Node<K,V>[] tab; int sc;
while ((tab = table) == null || tab.length == 0) {
//sizeCtl,代表着初始化资源或者扩容资源的锁,必须要获取到该锁才允许进行初始化或者扩容的操作
if ((sc = sizeCtl) < 0)
//放弃当前cpu的使用权,让出时间片,线程计入就绪状态参与竞争
Thread.yield(); // lost initialization race; just spin
//比较并尝试将sizeCtl替换成-1,如果失败则继续循环
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
try {
//进行一次double check 防止在进入分支前,容器发生了变更
if ((tab = table) == null || tab.length == 0) {
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;
}
/**
* 计算成员数量
*/
private final void addCount(long x, int check) {
CounterCell[] as; long b, s;
//统计单元不为空or数量增加失败
if ((as = counterCells) != null ||
!U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
CounterCell a; long v; int m;
boolean uncontended = true;
if (as == null || (m = as.length - 1) < 0 ||
(a = as[ThreadLocalRandom.getProbe() & m]) == null ||
!(uncontended =
U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {
//创建cells或者进行数量增加
fullAddCount(x, uncontended);
return;
}
if (check <= 1)
return;
s = sumCount();
}
//检查是否需要扩容
if (check >= 0) {
Node<K,V>[] tab, nt; int n, sc;
while (s >= (long)(sc = sizeCtl) && (tab = table) != null && (n = tab.length) < MAXIMUM_CAPACITY) {
int rs = resizeStamp(n);
if (sc < 0) {
if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
transferIndex <= 0)
break;
if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
transfer(tab, nt);
}
else if (U.compareAndSwapInt(this, SIZECTL, sc,(rs << RESIZE_STAMP_SHIFT) + 2))
transfer(tab, null);
s = sumCount();
}
}
}
/**
* 进行扩容
*/
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
int n = tab.length, stride;
//【第一步】
//决定当前线程在需要处理的槽位充足下,分配到的槽位数
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
stride = MIN_TRANSFER_STRIDE; // subdivide range
//新容器为空则创建容器
if (nextTab == null) { // initiating
try {
//多出一个赋值操作,尝试处理内存溢出?不明白原理
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];
nextTab = nt;
} catch (Throwable ex) { // try to cope with OOME
sizeCtl = Integer.MAX_VALUE;
return;
}
nextTable = nextTab;
//转移索引数设置为当前容器容量
transferIndex = n;
}
//将下个容器的转移搜索引数设置为新容器容量
int nextn = nextTab.length;
//创建ForwardingNode容器并放入新容器
ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
boolean advance = true;
boolean finishing = false; // to ensure sweep before committing nextTab
for (int i = 0, bound = 0;;) {
Node<K,V> f; int fh;
//【第二步,划分槽位,帮助推进】
//选择当前线程进行transfer的槽位,从最后一个槽位向前
while (advance) {
int nextIndex, nextBound;
//向前推进一个槽位,或者已经完成了
if (--i >= bound || finishing)
advance = false;
//槽位被其它线程选择完了
else if ((nextIndex = transferIndex) <= 0) {
i = -1;
advance = false;
}
//尝试获取槽位的操作权
else if (U.compareAndSwapInt
(this, TRANSFERINDEX, nextIndex,
nextBound = (nextIndex > stride ?
nextIndex - stride : 0))) {
//槽位下限
bound = nextBound;
//当前选中进行处理的槽位
i = nextIndex - 1;
advance = false;
}
}
//被选择完毕,选中槽位大于当前容器容量,选中槽位+当前容器容量大于新容器容量
//【第三步,设置结束条件,变更地址】
if (i < 0 || i >= n || i + n >= nextn) {
int sc;
//扩容完毕
if (finishing) {
//清除扩容时创建的临时表
nextTable = null;
//将当前表指向临时表
table = nextTab;
//设置下次扩容的临界点为 0.75*扩容容量
sizeCtl = (n << 1) - (n >>> 1);
return;
}
//将扩容标识中的线程标识减一
if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
//存在其它线程进行扩容处理,则当前线程处理完自己的槽位后直接退出
if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
return;
//不存在其它线程处理,说明自己是唯一处理线程
finishing = advance = true;
//将i重置,在看下还有没有transferIndex
//如果已经是唯一处理线程并且满足前置条件,为何需要检查下?
i = n; // recheck before commit
}
}
//【第四步,处理槽位】
//如果当前槽中没有成员,用forwarding节点占位
else if ((f = tabAt(tab, i)) == null)
advance = casTabAt(tab, i, null, fwd);
//如果当前槽中成员为forwarding节点,代表已经被处理过了
else if ((fh = f.hash) == MOVED)
//处理下一个槽
advance = true; // already processed
else {
//锁住槽位
synchronized (f) {
//double check
if (tabAt(tab, i) == f) {
Node<K,V> ln, hn;
if (fh >= 0) {
//计算当前成员最高位
//runBit是0 or 1
int runBit = fh & n;
Node<K,V> lastRun = f;
for (Node<K,V> p = f.next; p != null; p = p.next) {
int b = p.hash & n;
//查找最后重复的链,获得开始位置p,和重复的高位值runBit
if (b != runBit) {
runBit = b;
lastRun = p;
}
}
//如果从p开始后面高位全是0,那么就不需要移动到新槽中
if (runBit == 0) {
ln = lastRun;
hn = null;
}
//如果从p开始后面全是1,那么就需要移动到新槽中
else {
hn = lastRun;
ln = null;
}
//从链的头部一直遍历到p的位置(因为p以后高位都一样)
//为何需要提前找一部分重复?效率更高?这么处理是否有理论依据?
for (Node<K,V> p = f; p != lastRun; p = p.next) {
int ph = p.hash; K pk = p.key; V pv = p.val;
//高位为0放到旧槽位中
if ((ph & n) == 0)
ln = new Node<K,V>(ph, pk, pv, ln);
//高位为1放到新槽位中
else
hn = new Node<K,V>(ph, pk, pv, hn);
}
//将ln放到新容器的旧槽位中
setTabAt(nextTab, i, ln);
//将hn放到新容器的新槽位中
setTabAt(nextTab, i + n, hn);
//将老容器中的该节点设置为forwarding节点
setTabAt(tab, i, fwd);
//处理下一个槽位
advance = true;
}
//TreeBin的hash固定为-2,红黑树的调整
else if (f instanceof TreeBin) {
TreeBin<K,V> t = (TreeBin<K,V>)f;
TreeNode<K,V> lo = null, loTail = null;
TreeNode<K,V> hi = null, hiTail = null;
int lc = 0, hc = 0;
for (Node<K,V> e = t.first; e != null; e = e.next) {
int h = e.hash;
TreeNode<K,V> p = new TreeNode<K,V>
(h, e.key, e.val, null, null);
if ((h & n) == 0) {
if ((p.prev = loTail) == null)
lo = p;
else
loTail.next = p;
loTail = p;
++lc;
}
else {
if ((p.prev = hiTail) == null)
hi = p;
else
hiTail.next = p;
hiTail = p;
++hc;
}
}
//槽位里成员少于等于6,退化为链表
ln = (lc <= UNTREEIFY_THRESHOLD) ? untreeify(lo) :
(hc != 0) ? new TreeBin<K,V>(lo) : t;
hn = (hc <= UNTREEIFY_THRESHOLD) ? untreeify(hi) :
(lc != 0) ? new TreeBin<K,V>(hi) : t;
setTabAt(nextTab, i, ln);
setTabAt(nextTab, i + n, hn);
setTabAt(tab, i, fwd);
advance = true;
}
}
}
}
}
}
2.get 方法
不加锁
1.根据 key 计算出 hash 值,判断数组是否为空;
2.如果是首节点,就直接返回;
3.如果是红黑树结构,就从红黑树里面查询;
4.如果是链表结构,循环遍历判断。
public V get(Object key) {
// tab 引用map.table、 e 当前元素、 p 目标结点、 n 长度、 eh 当前元素的hash、 ek 当前元素的key
Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
int h = spread(key.hashCode());// 通过扰动运算后得到 更散列的hash值
// 表已经创建了 而且头结点不等于null
if ((tab = table) != null && (n = tab.length) > 0 &&
(e = tabAt(tab, (n - 1) & h)) != null) {
// 头结点直接找到
if ((eh = e.hash) == h) {
if ((ek = e.key) == key || (ek != null && key.equals(ek)))
return e.val;
}
// -1 fwd结点说明table正在扩容 且当前查询的已经被迁移走了
// -2 Treebin 需要使用Treebin方法查询
else if (eh < 0)
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、ConcurrentSkipListMap(代替TreeMap)
使用红黑树按照key的顺序(自然顺序、自定义顺序)来使得键值对有序存储的底层是通过跳表来实现的。
3、ConcurrentSkipListSet(代替TreeSet)
基于 ConcurrentSkipListMap 实现的,ConcurrentSkipListMap的键就不重复。
4、ConcurrentLinkedQueue(BlockingQueue)
通过无锁的方式,实现了高并发状态下的高性能,通常ConcurrentLikedQueue性能好于BlockingQueue。