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());
                }
            }
        }

• 总结
  • 了解数据结构并发独写的问题
  • 根据业务特点, 使用正确的并发数据结构