devbox
盐析白兔
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Java 提供的线程安全集合_java 线程安全的数组

作者：盐析白兔 | 2024-05-04 16:02:40

赞

踩

java 线程安全的数组

一、CopyOnWrite（COW算法的容器）

最终一致性、写分离思想。
用Volatile修饰，每次直接从内存地址中读取，读取时不加锁。
写时用显式锁整个容器（防止其它写线程），然后拷贝一份副本，对副本操作，读线程访问原容器数据。
内存开销大，实时数据一致性不高。适用于读远大于写操作且数据量不大的场景。

1、CopyOnWriteArrayList

底层是通过数组来实现的

1.构造方法

  1. /**
  2.  * 无参构造方法
  3.  */
  4. public CopyOnWriteArrayList() {
  5.     setArray(new Object[0]);
  6. }
  7. final void setArray(Object[] a) {
  8.     array = a;
  9. }
  10. final Object[] getArray() {
  11.         return array;
  12. }
  13. /**
  14.  * 传入集合构造方法
  15.  */
  16. public CopyOnWriteArrayList(Collection<? extends E> c) {
  17.     Object[] elements;
  18.     if (c.getClass() == CopyOnWriteArrayList.class){
  19.         elements = ((CopyOnWriteArrayList<?>)c).getArray();
  20.     } else {
  21.         elements = c.toArray();
  22.         if (c.getClass() != ArrayList.class)
  23.             elements = Arrays.copyOf(elements, elements.length, Object[].class);
  24.     }
  25.     setArray(elements);
  26. }
  27. /**
  28.  * 传入数组构造方法
  29.  */
  30. public CopyOnWriteArrayList(E[] toCopyIn) {
  31.     setArray(Arrays.copyOf(toCopyIn, toCopyIn.length, Object[].class));
  32. }

2.add、set 方法

  1. /**
  2.  * 新增元素
  3.  */
  4. public boolean add(E e) {
  5.     // 获取当前实例的可重入锁
  6.     final ReentrantLock lock = this.lock;
  7.     // 阻塞式加锁
  8.     lock.lock();
  9.     try {
  10.         // 获取当前集合的底层数组
  11.         Object[] elements = getArray();
  12.         int len = elements.length;
  13.         // 将当前数组的全部数据拷贝到新数组中，新数组长度为当前长度+1
  14.         Object[] newElements = Arrays.copyOf(elements, len + 1);
  15.         // 设置新增数据到数组中
  16.         newElements[len] = e;
  17.         // 将当前实例的底层数组指向新数组
  18.         setArray(newElements);
  19.         return true;
  20.     } finally {
  21.         lock.unlock();
  22.     }
  23. }
  24.  
  25. /**
  26.  * 修改元素
  27.  */
  28. public E set(int index, E element) {
  29.     // 获取锁
  30.     final ReentrantLock lock = this.lock;
  31.     lock.lock();
  32.     try {
  33.         Object[] elements = getArray();
  34.         // 获取源数组中该索引的值
  35.         E oldValue = get(elements, index);
  36.         // 如果需要设置的值与该索引值不一致则进行设置
  37.         if (oldValue != element) {
  38.             int len = elements.length;
  39.             // 拷贝原数组数据到新的数组
  40.             Object[] newElements = Arrays.copyOf(elements, len);
  41.             // 设置索引位置的元素值
  42.             newElements[index] = element;
  43.             setArray(newElements);
  44.         } else {
  45.             // Not quite a no-op; ensures volatile write semantics
  46.             setArray(elements);
  47.         }
  48.         return oldValue;
  49.     } finally {
  50.         lock.unlock();
  51.     }
  52. }

2.remove方法

  1. public E remove(int index) {
  2.     // 获取锁
  3.     final ReentrantLock lock = this.lock;
  4.     lock.lock();
  5.     try {
  6.         Object[] elements = getArray();
  7.         int len = elements.length;
  8.         // 从数组中获取需要删除的数据
  9.         E oldValue = get(elements, index);
  10.         // 计算需要移动元素的索引值
  11.         int numMoved = len - index - 1;
  12.         if (numMoved == 0)
  13.             // 需要移动元素的索引值为0表示待删除数据为最后一个元素，拷贝至新数组即可
  14.             setArray(Arrays.copyOf(elements, len - 1));
  15.         else {
  16.             // 需要移动的索引值大于0则需要分段拷贝至新数组
  17.             Object[] newElements = new Object[len - 1];
  18.             System.arraycopy(elements, 0, newElements, 0, index);
  19.             System.arraycopy(elements, index + 1, newElements, index, numMoved);
  20.             setArray(newElements);
  21.         }
  22.         return oldValue;
  23.     } finally {
  24.         lock.unlock();
  25.     }
  26. }

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 的时候，数组扩容或者将链表转换为红黑树。

  1. public V put(K key, V value) {
  2.  return putVal(key, value, false);
  3. }
  4. final V putVal(K key, V value, boolean onlyIfAbsent) {
  5.     if (key == null || value == null) throw new NullPointerException();
  6.     //获取hash值
  7.     int hash = spread(key.hashCode());
  8.     int binCount = 0;
  9.     for (Node<K,V>[] tab = table;;) {
  10.         Node<K,V> f; int n, i, fh;
  11.         //容器为空进行初始化流程
  12.         if (tab == null || (n = tab.length) == 0)
  13.             tab = initTable();
  14.         //如果槽位中为空的
  15.         else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
  16.             //以cas方式进行替换，替换成功就中断循环，替换失败则进行下一次循环
  17.             if (casTabAt(tab, i, null,new Node<K,V>(hash, key, value, null)))
  18.                 break;                   // no lock when adding to empty bin
  19.         }
  20.         //一种特殊的节点（forwarding 节点，迁移节点，只在迁移过程中存在）的处理方式
  21.         else if ((fh = f.hash) == MOVED)
  22.             //帮助进行扩容
  23.             tab = helpTransfer(tab, f);
  24.         //如果槽位不为空，并且不是（forwarding节点）
  25.         else {
  26.             V oldVal = null;
  27.             //将整个槽位锁住
  28.             synchronized (f) {
  29.                 //double check，如果槽位里面的数据发生变更则重新走流程
  30.                 if (tabAt(tab, i) == f) {
  31.                     //如果是链表
  32.                     if (fh >= 0) {
  33.                         binCount = 1;
  34.                         for (Node<K,V> e = f;; ++binCount) {
  35.                             K ek;
  36.                             //如果存在成员则覆盖
  37.                             if (e.hash == hash &&
  38.                                 ((ek = e.key) == key ||
  39.                                  (ek != null && key.equals(ek)))) {
  40.                                 oldVal = e.val;
  41.                                 if (!onlyIfAbsent)
  42.                                     e.val = value;
  43.                                 break;
  44.                             }
  45.                             //不存在成员则新增
  46.                             Node<K,V> pred = e;
  47.                             if ((e = e.next) == null) {
  48.                                 pred.next = new Node<K,V>(hash, key, value, null);
  49.                                 break;
  50.                             }
  51.                         }
  52.                     }
  53.                     //如果是红黑树
  54.                     else if (f instanceof TreeBin) {
  55.                         Node<K,V> p;
  56.                         binCount = 2;
  57.                         if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
  58.                                                        value)) != null) {
  59.                             oldVal = p.val;
  60.                             if (!onlyIfAbsent)
  61.                                 p.val = value;
  62.                         }
  63.                     }
  64.                 }
  65.             }
  66.             //如果一个槽位中的数量大于1（只有大于1的才会有binCount）
  67.             if (binCount != 0) {
  68.                 //如果槽位中的成员数量大于等于8，则变更为
  69.                 if (binCount >= TREEIFY_THRESHOLD)
  70.                     //进行转换成红黑树处理
  71.                     treeifyBin(tab, i);
  72.                 if (oldVal != null)
  73.                     return oldVal;
  74.                 break;
  75.             }
  76.         }
  77.     }
  78.     addCount(1L, binCount);
  79.     return null;
  80. }
  81.  
  82. /**
  83.  * 获取槽位 
  84. */
  85. static final int spread(int h) {
  86. 	//h为key值得hash值,将高16位也参与运算，然后与int最大值进行&运算(效果为将值变为正数，其他位置不变)
  87. 	//HASH_BITS为int最大值，最高位为0
  88. 	//HashMap中没有处理为正数的步骤，这里负数有其它含义，查看节点类型
  89. 	return (h ^ (h >>> 16)) & HASH_BITS;
  90. }
  91.  
  92. /**
  93.  * 初始化容器
  94. */
  95. private final Node<K,V>[] initTable() {
  96.    Node<K,V>[] tab; int sc;
  97.    while ((tab = table) == null || tab.length == 0) {
  98. 	   //sizeCtl，代表着初始化资源或者扩容资源的锁，必须要获取到该锁才允许进行初始化或者扩容的操作
  99. 	   if ((sc = sizeCtl) < 0)
  100. 		   //放弃当前cpu的使用权，让出时间片，线程计入就绪状态参与竞争
  101. 		   Thread.yield(); // lost initialization race; just spin
  102. 	   //比较并尝试将sizeCtl替换成-1，如果失败则继续循环    
  103. 	   else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
  104. 		   try {
  105. 			   //进行一次double check 防止在进入分支前，容器发生了变更
  106. 			   if ((tab = table) == null || tab.length == 0) {
  107. 				   int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
  108. 				   @SuppressWarnings("unchecked")
  109. 				   //初始化容器
  110. 				   Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
  111. 				   table = tab = nt;
  112. 				   sc = n - (n >>> 2);
  113. 			   }
  114. 		   } finally {
  115. 			   sizeCtl = sc;
  116. 		   }
  117. 		   break;
  118. 	   }
  119.    }
  120.    return tab;
  121. }
  122.  
  123. /**
  124.  * 计算成员数量
  125. */
  126. private final void addCount(long x, int check) {
  127. 	CounterCell[] as; long b, s;
  128. 	//统计单元不为空or数量增加失败
  129. 	if ((as = counterCells) != null ||
  130. 		!U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
  131. 		CounterCell a; long v; int m;
  132. 		boolean uncontended = true;
  133. 		if (as == null || (m = as.length - 1) < 0 ||
  134. 			(a = as[ThreadLocalRandom.getProbe() & m]) == null ||
  135. 			!(uncontended =
  136. 			  U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {
  137. 			//创建cells或者进行数量增加
  138. 			fullAddCount(x, uncontended);
  139. 			return;
  140. 		}
  141. 		if (check <= 1)
  142. 			return;
  143. 		s = sumCount();
  144. 	}
  145. 	//检查是否需要扩容
  146. 	if (check >= 0) {
  147. 		Node<K,V>[] tab, nt; int n, sc;
  148. 		while (s >= (long)(sc = sizeCtl) && (tab = table) != null && (n = tab.length) < MAXIMUM_CAPACITY) {
  149. 			int rs = resizeStamp(n);
  150. 			if (sc < 0) {
  151. 				if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
  152. 					sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
  153. 					transferIndex <= 0)
  154. 					break;
  155. 				if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
  156. 					transfer(tab, nt);
  157. 			}
  158. 			else if (U.compareAndSwapInt(this, SIZECTL, sc,(rs << RESIZE_STAMP_SHIFT) + 2))
  159. 				transfer(tab, null);
  160. 			s = sumCount();
  161. 		}
  162. 	}
  163. }
  164.  
  165. /**
  166.  * 进行扩容
  167. */
  168. private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
  169. 	int n = tab.length, stride;
  170. 	//【第一步】
  171. 	//决定当前线程在需要处理的槽位充足下，分配到的槽位数
  172. 	if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
  173. 		stride = MIN_TRANSFER_STRIDE; // subdivide range
  174. 	//新容器为空则创建容器    
  175. 	if (nextTab == null) {            // initiating
  176. 		try {
  177. 			//多出一个赋值操作，尝试处理内存溢出？不明白原理
  178. 			@SuppressWarnings("unchecked")
  179. 			Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];
  180. 			nextTab = nt;
  181. 		} catch (Throwable ex) {      // try to cope with OOME
  182. 			sizeCtl = Integer.MAX_VALUE;
  183. 			return;
  184. 		}
  185. 		nextTable = nextTab;
  186. 		//转移索引数设置为当前容器容量
  187. 		transferIndex = n;
  188. 	}
  189. 	//将下个容器的转移搜索引数设置为新容器容量
  190. 	int nextn = nextTab.length;
  191. 	//创建ForwardingNode容器并放入新容器
  192. 	ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
  193. 	boolean advance = true;
  194. 	boolean finishing = false; // to ensure sweep before committing nextTab
  195. 	for (int i = 0, bound = 0;;) {
  196. 		Node<K,V> f; int fh;
  197. 		//【第二步，划分槽位，帮助推进】
  198. 		//选择当前线程进行transfer的槽位，从最后一个槽位向前
  199. 		while (advance) {
  200. 			int nextIndex, nextBound;
  201. 			//向前推进一个槽位，或者已经完成了
  202. 			if (--i >= bound || finishing)
  203. 				advance = false;
  204. 			//槽位被其它线程选择完了    
  205. 			else if ((nextIndex = transferIndex) <= 0) {
  206. 				i = -1;
  207. 				advance = false;
  208. 			}
  209. 			//尝试获取槽位的操作权
  210. 			else if (U.compareAndSwapInt
  211. 					 (this, TRANSFERINDEX, nextIndex,
  212. 					  nextBound = (nextIndex > stride ?
  213. 								   nextIndex - stride : 0))) {
  214. 				//槽位下限                   
  215. 				bound = nextBound;
  216. 				//当前选中进行处理的槽位
  217. 				i = nextIndex - 1;
  218. 				advance = false;
  219. 			}
  220. 		}
  221. 		//被选择完毕，选中槽位大于当前容器容量，选中槽位+当前容器容量大于新容器容量
  222. 		//【第三步，设置结束条件，变更地址】
  223. 		if (i < 0 || i >= n || i + n >= nextn) {
  224. 			int sc;
  225. 			//扩容完毕
  226. 			if (finishing) {
  227. 				//清除扩容时创建的临时表
  228. 				nextTable = null;
  229. 				//将当前表指向临时表
  230. 				table = nextTab;
  231. 				//设置下次扩容的临界点为 0.75*扩容容量
  232. 				sizeCtl = (n << 1) - (n >>> 1);
  233. 				return;
  234. 			}
  235. 			//将扩容标识中的线程标识减一
  236. 			if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
  237. 				//存在其它线程进行扩容处理，则当前线程处理完自己的槽位后直接退出
  238. 				if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
  239. 					return;
  240. 				//不存在其它线程处理，说明自己是唯一处理线程   
  241. 				finishing = advance = true;
  242. 				//将i重置，在看下还有没有transferIndex
  243. 				//如果已经是唯一处理线程并且满足前置条件，为何需要检查下？
  244. 				i = n; // recheck before commit
  245. 			}
  246. 		}
  247. 		//【第四步，处理槽位】
  248. 		//如果当前槽中没有成员，用forwarding节点占位
  249. 		else if ((f = tabAt(tab, i)) == null)
  250. 			advance = casTabAt(tab, i, null, fwd);
  251. 		//如果当前槽中成员为forwarding节点，代表已经被处理过了    
  252. 		else if ((fh = f.hash) == MOVED)
  253. 			//处理下一个槽
  254. 			advance = true; // already processed
  255. 		else {
  256. 			//锁住槽位
  257. 			synchronized (f) {
  258. 				//double check
  259. 				if (tabAt(tab, i) == f) {
  260. 					Node<K,V> ln, hn;
  261. 					if (fh >= 0) {
  262. 						//计算当前成员最高位
  263. 						//runBit是0 or 1
  264. 						int runBit = fh & n;
  265. 						Node<K,V> lastRun = f;
  266. 						for (Node<K,V> p = f.next; p != null; p = p.next) {
  267. 							int b = p.hash & n;
  268. 							//查找最后重复的链，获得开始位置p，和重复的高位值runBit
  269. 							if (b != runBit) {
  270. 								runBit = b;
  271. 								lastRun = p;
  272. 							}
  273. 						}
  274. 						//如果从p开始后面高位全是0，那么就不需要移动到新槽中
  275. 						if (runBit == 0) {
  276. 							ln = lastRun;
  277. 							hn = null;
  278. 						}
  279. 						//如果从p开始后面全是1，那么就需要移动到新槽中
  280. 						else {
  281. 							hn = lastRun;
  282. 							ln = null;
  283. 						}
  284. 						//从链的头部一直遍历到p的位置（因为p以后高位都一样）
  285. 						//为何需要提前找一部分重复？效率更高？这么处理是否有理论依据？
  286. 						for (Node<K,V> p = f; p != lastRun; p = p.next) {
  287. 							int ph = p.hash; K pk = p.key; V pv = p.val;
  288. 							//高位为0放到旧槽位中
  289. 							if ((ph & n) == 0)
  290. 								ln = new Node<K,V>(ph, pk, pv, ln);
  291. 							//高位为1放到新槽位中
  292. 							else
  293. 								hn = new Node<K,V>(ph, pk, pv, hn);
  294. 						}
  295. 						//将ln放到新容器的旧槽位中
  296. 						setTabAt(nextTab, i, ln);
  297. 						//将hn放到新容器的新槽位中
  298. 						setTabAt(nextTab, i + n, hn);
  299. 						//将老容器中的该节点设置为forwarding节点
  300. 						setTabAt(tab, i, fwd);
  301. 						//处理下一个槽位
  302. 						advance = true;
  303. 					}
  304. 					//TreeBin的hash固定为-2，红黑树的调整
  305. 					else if (f instanceof TreeBin) {
  306. 						TreeBin<K,V> t = (TreeBin<K,V>)f;
  307. 						TreeNode<K,V> lo = null, loTail = null;
  308. 						TreeNode<K,V> hi = null, hiTail = null;
  309. 						int lc = 0, hc = 0;
  310. 						for (Node<K,V> e = t.first; e != null; e = e.next) {
  311. 							int h = e.hash;
  312. 							TreeNode<K,V> p = new TreeNode<K,V>
  313. 								(h, e.key, e.val, null, null);
  314. 							if ((h & n) == 0) {
  315. 								if ((p.prev = loTail) == null)
  316. 									lo = p;
  317. 								else
  318. 									loTail.next = p;
  319. 								loTail = p;
  320. 								++lc;
  321. 							}
  322. 							else {
  323. 								if ((p.prev = hiTail) == null)
  324. 									hi = p;
  325. 								else
  326. 									hiTail.next = p;
  327. 								hiTail = p;
  328. 								++hc;
  329. 							}
  330. 						}
  331. 						//槽位里成员少于等于6，退化为链表
  332. 						ln = (lc <= UNTREEIFY_THRESHOLD) ? untreeify(lo) :
  333. 							(hc != 0) ? new TreeBin<K,V>(lo) : t;
  334. 						hn = (hc <= UNTREEIFY_THRESHOLD) ? untreeify(hi) :
  335. 							(lc != 0) ? new TreeBin<K,V>(hi) : t;
  336. 						setTabAt(nextTab, i, ln);
  337. 						setTabAt(nextTab, i + n, hn);
  338. 						setTabAt(tab, i, fwd);
  339. 						advance = true;
  340. 					}
  341. 				}
  342. 			}
  343. 		}
  344. 	}
  345. }

2.get 方法 

 不加锁
1.根据 key 计算出 hash 值，判断数组是否为空；
2.如果是首节点，就直接返回；
3.如果是红黑树结构，就从红黑树里面查询；
4.如果是链表结构，循环遍历判断。

  1. public V get(Object key) {
  2. 	// tab 引用map.table、 e 当前元素、 p 目标结点、 n 长度、 eh 当前元素的hash、 ek 当前元素的key
  3. 	Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
  4. 	int h = spread(key.hashCode());// 通过扰动运算后得到 更散列的hash值
  5. 	// 表已经创建了 而且头结点不等于null
  6. 	if ((tab = table) != null && (n = tab.length) > 0 &&
  7. 		(e = tabAt(tab, (n - 1) & h)) != null) {
  8. 		// 头结点直接找到
  9. 		if ((eh = e.hash) == h) {
  10. 			if ((ek = e.key) == key || (ek != null && key.equals(ek)))
  11. 				return e.val;
  12. 		}
  13. 		// -1 fwd结点说明table正在扩容 且当前查询的已经被迁移走了
  14. 		// -2 Treebin 需要使用Treebin方法查询
  15. 		else if (eh < 0)
  16. 			return (p = e.find(h, key)) != null ? p.val : null;
  17. 		while ((e = e.next) != null) {// 链表情况
  18. 			if (e.hash == h &&
  19. 				((ek = e.key) == key || (ek != null && key.equals(ek))))
  20. 				return e.val;
  21. 		}
  22. 	}
  23. 	return null;
  24. }

2、ConcurrentSkipListMap（代替TreeMap）

使用红黑树按照key的顺序（自然顺序、自定义顺序）来使得键值对有序存储的底层是通过跳表来实现的。

3、ConcurrentSkipListSet（代替TreeSet）

基于 ConcurrentSkipListMap 实现的,ConcurrentSkipListMap的键就不重复。

4、ConcurrentLinkedQueue（BlockingQueue）

通过无锁的方式，实现了高并发状态下的高性能，通常ConcurrentLikedQueue性能好于BlockingQueue。

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/盐析白兔/article/detail/535155
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号