Java 多线程进阶-并发数据结构
Posted Sweet小马
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Java 多线程进阶-并发数据结构相关的知识,希望对你有一定的参考价值。
Java 多线程进阶-并发数据结构
- 并发数据结构
- 常用的数据结构是线程不安全的
- ArrayList/HashMap/HashSet 非同步的
- 多个线程同时独写, 可能会抛出异常或数据错误
- 传统Vector/HashTable等同步数据集合性能过差
- 并发数据结构: 数据添加或删除
- 阻塞式集合: 当集合为空或者满时, 等待
- 非阻塞式集合: 当集合为空或者满时, 不等待, 返回null或异常
- List
- Vector 同步安全, 写多读少, 效率差
- ArrayList 不安全
- 通过 基于 synchronized的Collections.synchronizedList(List list) 方法将 List 变成线程安全的, 效率差
- CopyOnWriteArrayList (JDK5提供的基于复制机制的并发列表类), 非阻塞的, 适用于读多写少即填入数据之后大部分操作是读取和遍历的. 效率好, 适合在多线程下使用.
package thread0418; import java.time.LocalDateTime; import java.util.ArrayList; import java.util.Collections; import java.util.List; import java.util.concurrent.CopyOnWriteArrayList; /** * 并发数据结构对比 */ public class ThreadListDemo1 { public static void main(String[] args) throws InterruptedException { // 线程不安全 List<String> unsafeList = new ArrayList<String>(); // 线程安全 将一个不安全的转成安全的 List<String> safeList1 = Collections.synchronizedList(new ArrayList<String>()); // 线程安全 CopyOnWriteArrayList<String> safeList2 = new CopyOnWriteArrayList<String>(); ListThread111 t1 = new ListThread111(unsafeList); ListThread111 t2 = new ListThread111(safeList1); ListThread111 t3 = new ListThread111(safeList2); // 分别启动十个线程, 运行测试 for (int i = 0; i < 10; i++) { Thread t = new Thread(t1, String.valueOf(i)); t.start(); } for (int i = 0; i < 10; i++) { Thread t = new Thread(t2, String.valueOf(i)); t.start(); } for (int i = 0; i < 10; i++) { Thread t = new Thread(t3, String.valueOf(i)); t.start(); } // 等待子线程执行完 Thread.sleep(2000); System.out.println(LocalDateTime.now() + " => " + t1.list.size()); System.out.println(LocalDateTime.now() + " => " + t2.list.size()); System.out.println(LocalDateTime.now() + " => " + t3.list.size()); // 输出list中的值 System.out.println(LocalDateTime.now() + " => " + "unsafeList:"); for (String s : t1.list) { if (s == null) { System.out.print("null "); } else { System.out.print(s + " "); } } System.out.println(); System.out.println(LocalDateTime.now() + " => " + "safeList1: "); for (String s : t2.list) { if (s == null) { System.out.print("null "); } else { System.out.print(s + " "); } } System.out.println(); System.out.println(LocalDateTime.now() + " => " + "safeList2: "); for (String s : t3.list) { if (s == null) { System.out.print("null "); } else { System.out.print(s + " "); } } } } class ListThread111 implements Runnable { public List<String> list; public ListThread111(List<String> list) { this.list = list; } @Override public void run() { int i = 0; while (i < 10) { try { Thread.sleep(10); } catch (Exception e) { e.printStackTrace(); } // 把当前线程名字加到 list 中 list.add(Thread.currentThread().getName()); i++; } } }
- Set
- HashSet 不安全
- Collections.synchronizedSet(Set set) 基于 synchronized, 效率差
- CopyOnWriteArraySet (JDK5提供的基于 CopyOnWriteArrayList 实现), 非阻塞的, 适用于读多写少, 效率好, 适合在多线程下使用.
package thread0418; import java.time.LocalDateTime; import java.util.Collections; import java.util.HashSet; import java.util.Set; import java.util.concurrent.CopyOnWriteArraySet; public class ThreadSetDemo1 { public static void main(String[] args) throws InterruptedException { // 线程不安全 Set<String> unsafeSet = new HashSet<String>(); // 线程安全 Set<String> safeSet1 = Collections.synchronizedSet(new HashSet<>()); // 线程安全 CopyOnWriteArraySet<String> safeSet2 = new CopyOnWriteArraySet<>(); SetThread111 t1 = new SetThread111(unsafeSet); SetThread111 t2 = new SetThread111(safeSet1); SetThread111 t3 = new SetThread111(safeSet2); // 分别启动十个线程, 运行测试 for (int i = 0; i < 10; i++) { Thread t = new Thread(t1, String.valueOf(i)); t.start(); } for (int i = 0; i < 10; i++) { Thread t = new Thread(t2, String.valueOf(i)); t.start(); } for (int i = 0; i < 10; i++) { Thread t = new Thread(t3, String.valueOf(i)); t.start(); } // 等待子线程执行完成 Thread.sleep(2000); System.out.println(LocalDateTime.now() + " => " + t1.set.size()); System.out.println(LocalDateTime.now() + " => " + t2.set.size()); System.out.println(LocalDateTime.now() + " => " + t3.set.size()); // 输出 Set 中的值 System.out.println(LocalDateTime.now() + " => " + "unsafeSet: "); for (String ele : t1.set) { if (ele == null) { System.out.print("null "); } else { System.out.print(ele + " "); } } System.out.println(); System.out.println(LocalDateTime.now() + " => " + "safeSet1: "); for (String ele : t2.set) { if (ele == null) { System.out.print("null "); } else { System.out.print(ele + " "); } } System.out.println(); System.out.println(LocalDateTime.now() + " => " + "safeSet2: "); for (String ele : t3.set) { if (ele == null) { System.out.print("null "); } else { System.out.print(ele + " "); } } } } class SetThread111 implements Runnable { public Set<String> set; public SetThread111(Set<String> set) { this.set = set; } @Override public void run() { int i = 0; while (i < 10) { try { Thread.sleep(10); } catch (InterruptedException e) { e.printStackTrace(); } // 把当前线程名称加入 set里 set.add(Thread.currentThread().getName() + i); i++; } } }
- Map
- Hashtable 同步安全, 写多读少, 效率差
- HashMap 不安全
- Collections.synchronizedMap(Map map) 基于 synchronized, 效率差
- ConcurrentHashMap 读多写少, 非阻塞
package thread0418; import java.time.LocalDateTime; import java.util.Collections; import java.util.HashMap; import java.util.Iterator; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; public class ThreadMapDemo1 { public static void main(String[] args) throws InterruptedException { // 线程不安全 MapThread111 t1 = new MapThread111(new HashMap<>()); // 线程安全 MapThread111 t2 = new MapThread111(Collections.synchronizedMap(new HashMap<>())); // 线程安全 MapThread111 t3 = new MapThread111(new ConcurrentHashMap<>()); // 分别启动十个线程, 运行测试 for (int i = 0; i < 10; i++) { Thread t = new Thread(t1); t.start(); } for (int i = 0; i < 10; i++) { Thread t = new Thread(t2); t.start(); } for (int i = 0; i < 10; i++) { Thread t = new Thread(t3); t.start(); } // 等待子线程执行完 Thread.sleep(2000); System.out.println(LocalDateTime.now() + " => " + t1.map.size()); System.out.println(LocalDateTime.now() + " => " + t2.map.size()); System.out.println(LocalDateTime.now() + " => " + t3.map.size()); // 输出 map 中的值 System.out.println(LocalDateTime.now() + " => " + "unsafeMap:"); Iterator<Map.Entry<Integer, String>> iterator = t1.map.entrySet().iterator(); while (iterator.hasNext()) { Map.Entry<Integer, String> entry = iterator.next(); System.out.print(entry.getKey() + ":" + entry.getValue() + " "); } System.out.println(); System.out.println(LocalDateTime.now() + " => " + "safeMap:"); iterator = t2.map.entrySet().iterator(); while (iterator.hasNext()) { Map.Entry<Integer, String> entry = iterator.next(); System.out.print(entry.getKey() + ":" + entry.getValue() + " "); } System.out.println(); System.out.println(LocalDateTime.now() + " => " + "safeMap2:"); iterator = t3.map.entrySet().iterator(); while (iterator.hasNext()) { Map.Entry<Integer, String> entry = iterator.next(); System.out.print(entry.getKey() + ":" + entry.getValue() + " "); } System.out.println(); System.out.println(LocalDateTime.now() + " => " + "mapThread1.map.size()" + t1.map.size()); System.out.println(LocalDateTime.now() + " => " + "mapThread2.map.size()" + t2.map.size()); System.out.println(LocalDateTime.now() + " => " + "mapThread3.map.size()" + t3.map.size()); } } class MapThread111 implements Runnable { public Map<Integer, String> map; public MapThread111(Map<Integer, String> map) { this.map = map; } @Override public void run() { int i = 0; while (i < 100) { try { Thread.sleep(10); } catch (InterruptedException e) { e.printStackTrace(); } // 把当前线程名称放入map中 map.put(++i, Thread.currentThread().getName()); } } }
- Queue(单向队列) & Deque(双向队列)(JDK1.5)
- ConcurrentLinkedQueue 非阻塞
- ArrayBlockingQueue/LinkedBlockingQueue 阻塞
package thread0418; import java.time.LocalDateTime; import java.util.ArrayDeque; import java.util.Queue; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.ConcurrentLinkedDeque; public class ThreadQueueDemo1 { public static void main(String[] args) throws InterruptedException { // 线程不安全 QueueThread111 t1 = new QueueThread111(new ArrayDeque<>()); // 线程安全 QueueThread111 t2 = new QueueThread111(new ConcurrentLinkedDeque<>()); QueueThread111 t3 = new QueueThread111(new ArrayBlockingQueue<>(100)); for (int i = 0; i < 10; i++) { Thread thread = new Thread(t1, String.valueOf(i)); thread.start(); } for (int i = 0; i < 10; i++) { Thread thread = new Thread(t2, String.valueOf(i)); thread.start(); } for (int i = 0; i < 10; i++) { Thread thread = new Thread(t3, String.valueOf(i)); thread.start(); } // 等待子线程执行完 Thread.sleep(2000); System.out.println(LocalDateTime.now() + " => " + t1.queue.size()); System.out.println(LocalDateTime.now() + " => " + t2.queue.size()); System.out.println(LocalDateTime.now() + " => " + t3.queue.size()); // 输出 Queue 中的值 System.out.println("unsafeQueue: "); for (String s : t1.queue) { System.out.print(s + " "); } System.out.println(); System.out.println("safeQueue1: "); for (String s : t2.queue) { System.out.print(s + " "); } System.out.println(); System.out.println("safeQueue2: "); for (String s : t3.queue) { System.out.print(s + " "); } } } class QueueThread111 implements Runnable { public Queue<String> queue; public QueueThread111(Queue<String> queue) { this.queue = queue; } @Override public void run() { int i = 0; while (i < 10) { i++; try { Thread.sleep(10); } catch (InterruptedException e) { e.printStackTrace(); } // 把当前线程名称加入list中 queue.add(Thread.currentThread().getName()); } } }
- 常用的数据结构是线程不安全的
- 总结
- 了解数据结构并发独写的问题
- 根据业务特点, 使用正确的并发数据结构
以上是关于Java 多线程进阶-并发数据结构的主要内容,如果未能解决你的问题,请参考以下文章