如何理解Java同步容器

169次閱讀

沒有評論

共計 9220 個字符，預計需要花費 24 分鐘才能閱讀完成。

如何理解 Java 同步容器，針對這個問題，這篇文章詳細介紹了相對應的分析和解答，希望可以幫助更多想解決這個問題的小伙伴找到更簡單易行的方法。

ArrayList,HashSet,HashMap 都是線程非安全的, 在多線程環境下, 會導致線程安全問題, 所以在使用的時候需要進行同步, 這無疑增加了程序開發的難度。所以 JAVA 提供了同步容器。

同步容器

ArrayList === Vector,Stack

HashMap === HashTable(key,value 都不能為空)

Collections.synchronizedXXX(List,Set,Map)

Vector 實現 List 接口，底層和 ArrayList 類似，但是 Vector 中的方法都是使用 synchronized 修飾，即進行了同步的措施。但是，Vector 并不是線程安全的。

Stack 也是一個同步容器，也是使用 synchronized 進行同步，繼承與 Vector，是數據結構中的，先進后出。

HashTable 和 HashMap 很相似，但 HashTable 進行了同步處理。

Collections 工具類提供了大量的方法，比如對集合的排序、查找等常用的操作。同時也通過了相關了方法創建同步容器類

Vector

package com.rumenz.task;
import java.util.List;
import java.util.Vector;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
// 線程安全
public class VectorExample1 {
 public static Integer clientTotal=5000;
 public static Integer thradTotal=200;
 private static List integer  list=new Vector ();
 public static void main(String[] args) throws Exception{ ExecutorService executorService = Executors.newCachedThreadPool();
 final Semaphore semaphore=new Semaphore(thradTotal);
 final CountDownLatch countDownLatch=new CountDownLatch(clientTotal);
 for (int i = 0; i   clientTotal; i++) {
 final Integer j=i;
 executorService.execute(()- {
 try { semaphore.acquire();
 update(j);
 semaphore.release();
 }catch (Exception e){ e.printStackTrace();
 }
 countDownLatch.countDown();
 });
 
 }
 countDownLatch.await();
 executorService.shutdown();
 System.out.println(size: +list.size());
 }
 private static void update(Integer j) { list.add(j);
 }
}

同步容器不一定就線程安全

package com.rumenz.task;
import scala.collection.convert.impl.VectorStepperBase;
import java.util.List;
import java.util.Vector;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
// 線程不安全
public class VectorExample2 {
 public static Integer clientTotal=5000;
 public static Integer threadTotal=200;
 private static List integer  list=new Vector();
 public static void main(String[] args) { ExecutorService executorService = Executors.newCachedThreadPool();
 final Semaphore semaphore=new Semaphore(threadTotal);
 for (int i = 0; i   threadTotal; i++) { list.add(i);
 }
 for (int i = 0; i   clientTotal; i++) {
 try{ semaphore.acquire();
 executorService.execute(()- { for (int j = 0; j   list.size(); j++) { list.remove(j);
 }
 });
 executorService.execute(()- { for (int j = 0; j   list.size(); j++) { list.get(j);
 }
 });
 semaphore.release();
 }catch (Exception e){ e.printStackTrace();
 }
 
 }
 executorService.shutdown();
 }
}

運行報錯

Exception in thread  pool-1-thread-2  java.lang.ArrayIndexOutOfBoundsException: Array index out of range: 36
 at java.util.Vector.get(Vector.java:751)
 at com.rumenz.task.VectorExample2.lambda$main$1(VectorExample2.java:38)
 at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
 at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
 at java.lang.Thread.run(Thread.java:748)

原因分析

Vector 是線程同步容器,size(),get(),remove() 都是被 synchronized 修飾的, 為什么會有線程安全問題呢?

get() 拋出的異常肯定是 remove() 引起的,Vector 能同時保證同一時刻只有一個線程進入, 但是:

// 線程 1
 executorService.execute(()- { for (int j = 0; j   list.size(); j++) { list.remove(j);
 }
// 線程 2
executorService.execute(()- { for (int j = 0; j   list.size(); j++) { list.get(j);
 }
});

線程 1 和線程 2 都執行完 list.size(), 都等于 200, 并且 j =100

線程 1 執行 list.remove(100) 操作,

線程 2 執行 list.get(100) 就會拋出數組越界的異常。

同步容器雖然是線程安全的, 但是不代表在任何環境下都是線程安全的。

HashTable

線程安全,key,value 都不能為 null。在修改數據時鎖住整個 HashTable, 效率低下。初始 size=11。

package com.rumenz.task;
import java.util.Hashtable;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
// 線程安全
public class HashTableExample1 {
 public static Integer clientTotal=5000;
 public static Integer threadTotal=200;
 private static Map integer,integer  map=new Hashtable ();
 public static void main(String[] args) throws Exception { ExecutorService executorService = Executors.newCachedThreadPool();
 final Semaphore semaphore=new Semaphore(threadTotal);
 final CountDownLatch countDownLatch=new CountDownLatch(clientTotal);
 for (int i = 0; i   clientTotal; i++) {
 final Integer j=i;
 try{ semaphore.acquire();
 update(j);
 semaphore.release();
 }catch (Exception e){ e.printStackTrace();
 }
 countDownLatch.countDown();
 
 }
 countDownLatch.await();
 executorService.shutdown();
 System.out.println(size: +map.size());

 private static void update(Integer j) { map.put(j, j);
 }
//size:5000

Collections.synchronizedList 線程安全

package com.rumenz.task.Collections;
import com.google.common.collect.Lists;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
// 線程安全
public class synchronizedExample {
 public static Integer clientTotal=5000;
 public static Integer threadTotal=200;
 private static List integer  list=Collections.synchronizedList(Lists.newArrayList());
 public static void main(String[] args) throws Exception{ ExecutorService executorService = Executors.newCachedThreadPool();
 final Semaphore semaphore=new Semaphore(threadTotal);
 final CountDownLatch countDownLatch=new CountDownLatch(clientTotal);
 for (int i = 0; i   clientTotal; i++) {
 final Integer j=i;
 try{ semaphore.acquire();
 update(j);
 semaphore.release();
 }catch (Exception e){ e.printStackTrace();
 }
 countDownLatch.countDown();
 }
 countDownLatch.await();
 executorService.shutdown();
 System.out.println(size: +list.size());
 }
 private static void update(Integer j) { list.add(j);
 }
//size:5000

Collections.synchronizedSet 線程安全

package com.rumenz.task.Collections;
import com.google.common.collect.Lists;
import org.assertj.core.util.Sets;
import java.util.Collections;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
// 線程安全
public class synchronizedSetExample {
 public static Integer clientTotal=5000;
 public static Integer threadTotal=200;
 private static Set integer  set=Collections.synchronizedSet(Sets.newHashSet());
 public static void main(String[] args) throws Exception{ ExecutorService executorService = Executors.newCachedThreadPool();
 final Semaphore semaphore=new Semaphore(threadTotal);
 final CountDownLatch countDownLatch=new CountDownLatch(clientTotal);
 for (int i = 0; i   clientTotal; i++) {
 final Integer j=i;
 try{ semaphore.acquire();
 update(j);
 semaphore.release();
 }catch (Exception e){ e.printStackTrace();
 }
 countDownLatch.countDown();
 }
 countDownLatch.await();
 executorService.shutdown();
 System.out.println(size: +set.size());
 }
 private static void update(Integer j) { set.add(j);
 }
//size:5000

Collections.synchronizedMap 線程安全

package com.rumenz.task.Collections;
import org.assertj.core.util.Sets;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
public class synchronizedMapExample {
 public static Integer clientTotal=5000;
 public static Integer threadTotal=200;
 private static Map integer,integer  map=Collections.synchronizedMap(new HashMap ());
 public static void main(String[] args) throws Exception{ ExecutorService executorService = Executors.newCachedThreadPool();
 final Semaphore semaphore=new Semaphore(threadTotal);
 final CountDownLatch countDownLatch=new CountDownLatch(clientTotal);
 for (int i = 0; i   clientTotal; i++) {
 final Integer j=i;
 try{ semaphore.acquire();
 update(j);
 semaphore.release();
 }catch (Exception e){ e.printStackTrace();
 }
 countDownLatch.countDown();
 }
 countDownLatch.await();
 executorService.shutdown();
 System.out.println(size: +map.size());
 }
 private static void update(Integer j) { map.put(j, j);
 }
//size:5000

Collections.synchronizedXXX 在迭代的時候, 需要開發者自己加上線程鎖控制代碼, 因為在整個迭代過程中循環外面不加同步代碼, 在一次次迭代之間, 其他線程對于這個容器的 add 或者 remove 會影響整個迭代的預期效果, 這個時候需要在循環外面加上 synchronized(XXX)。

集合的刪除

如果在使用 foreach 或 iterator 進集合的遍歷，

盡量不要在操作的過程中進行 remove 等相關的更新操作。

如果非要進行操作，則可以在遍歷的過程中記錄需要操作元素的序號，

待遍歷結束后方可進行操作，讓這兩個動作分開進行

package com.rumenz.task;
import com.google.common.collect.Lists;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;

 private static List integer  list=Collections.synchronizedList(Lists.newArrayList());
 public static void main(String[] args) { list.add(1);
 list.add(2);
 list.add(3);
 list.add(4);
 //del1();
 //del2();
 del3();
 }
 private static void del3() { for(Integer i:list){ if(i==4){ list.remove(i);
 }
 }
 }
 //Exception in thread  main  java.util.ConcurrentModificationException
 private static void del2() { Iterator integer  iterator = list.iterator();
 while (iterator.hasNext()){ Integer i = iterator.next();
 if(i==4){ list.remove(i);
 }
 }
 }
 //Exception in thread  main  java.util.ConcurrentModificationException
 private static void del1() { for (int i = 0; i   list.size(); i++) { if(list.get(i)==4){ list.remove(i);
 }
 }
 }
}

在單線程會出現以上錯誤，在多線程情況下，并且集合時共享的，出現異常的概率會更大，需要特別的注意。解決方案是希望在 foreach 或 iterator 時，對要操作的元素進行標記，待循環結束之后，在執行相關操作。

同步容器采用 synchronized 進行同步, 因此執行的性能會受到影響, 并且同步容器也并不一定會做到線程安全。

關于如何理解 Java 同步容器問題的解答就分享到這里了，希望以上內容可以對大家有一定的幫助，如果你還有很多疑惑沒有解開，可以關注丸趣 TV 行業資訊頻道了解更多相關知識。

正文完