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 多线程进阶-并发数据结构的主要内容,如果未能解决你的问题,请参考以下文章

Java 多线程进阶-并发协作控制

Java 多线程进阶-并发编程 线程组ThreadGroup

Java进阶 - 并发(多线程基本概念)

Java进阶 - 并发(多线程锁类)

Java进阶 - 并发(关键字)

Java知识体系Java并发编程进阶,多线程和锁底层原理探究