Java 多线程之线程安全集合

2023-12-14 01:18:42

一、概述

集合关系图
本文主要关注线程安全的集合，如 List、Set、Queue、Map 等接口的线程安全的实现方式，有关集合基础知识请转到这里。所谓线程安全集合，就是在多线程环境中使用集合不会导致数据不一致和数据异常的集合。在 Java 中线程安全集现在基本都使用 java.util.concurrent 包下的类。

二、List 接口线程安全实现类

2.1 普通 List 变线程安全 List

要想普通集合能在多线程环境中使用，最简单的办法就是使用 Collections.synchronizedList 方法，将普通集合转换为线程安全集合。当然也可以手动加锁，如 synchronized。
如 ArrayList 本身不是线程安全的，要使他具有线程安全，可以使用 Collections.synchronizedList(new ArrayList<>()) 。接下来的方法与使用普通 List 并不区别。
```
List<Object> list = Collections.synchronizedList(new ArrayList<>());
// 添加元素
arrayList.add("aaa");
arrayList.add("bbb");
arrayList.add("ccc");
```

通过查看 Collections.synchronizedList 的源码片段，可以看出本质上就是使用 synchronized 加锁。

public class Collections {
    // 省略大部分代码 ...
    public static <T> List<T> synchronizedList(List<T> list) {
        return (list instanceof RandomAccess ?
                new SynchronizedRandomAccessList<>(list) :
                new SynchronizedList<>(list));
    }
    // 省略大部分代码 ...
    static class SynchronizedList<E>
        extends SynchronizedCollection<E>
        implements List<E> {
        public E get(int index) {
            synchronized (mutex) {return list.get(index);}
        }
        public E set(int index, E element) {
            synchronized (mutex) {return list.set(index, element);}
        }
        public void add(int index, E element) {
            synchronized (mutex) {list.add(index, element);}
        }
        public E remove(int index) {
            synchronized (mutex) {return list.remove(index);}
        }
        // 省略大部分代码 ...
    }
}

当然， Map 和 Set 类型也一样，同样可以使用 Collections 将其转成线程案例的。

        Map<Object, Object> map = Collections.synchronizedMap(new HashMap<>());
        Set<Object> set = Collections.synchronizedSet(new HashSet<>());

2.2 Vector

Vector 是线程安全的，他的实现方式是给每个方法加 synchronized。因为他是早期的实现方式，某些情况下效率较低，所以一般情况下不会选用。

        // 创建一个 Vector
        Vector<String> vector = new Vector<>();
        // 添加元素
        vector.add("aaa");
        vector.add("bbb");
        vector.add("ccc");

Vector 线程安全实现源码片段，可以看出是在每个方法上加 synchronized。

public class Vector<E>
    extends AbstractList<E>
    implements List<E>, RandomAccess, Cloneable, java.io.Serializable
{
    // 省略大部分代码 ...
    public synchronized boolean add(E e) {
        modCount++;
        ensureCapacityHelper(elementCount + 1);
        elementData[elementCount++] = e;
        return true;
    }    
    public synchronized E get(int index) {
        if (index >= elementCount)
            throw new ArrayIndexOutOfBoundsException(index);

        return elementData(index);
    }
    // 省略大部分代码 ...
}

Vector 还有一个字类 Stack 也是线程安全的。是一个后进先出（LIFO）的数据结构，它支持在栈顶进行插入、删除和查找操作。由于其同步特性，可能会导致在高并发场景下的性能相对较低。在现代Java编程中，更常见的做法是使用 LinkedList 或 ArrayDeque 等非线程安全的实现，并在需要时通过显式的同步手段保证线程安全。这样可以在需要时灵活地选择适合具体场景的同步策略。

2.3 CopyOnWriteArrayList

CopyOnWriteArrayList 是线程安全的，基于数组操作，效率高，他适用于读多写少场景下。

        // 创建一个 CopyOnWriteArrayList
        List<String> copyOnWriteArrayList = new CopyOnWriteArrayList<>();
        // 添加元素
        copyOnWriteArrayList.add("aaa");
        copyOnWriteArrayList.add("bbb");
        copyOnWriteArrayList.add("ccc");

从 CopyOnWriteArrayList 源码片段，可以看出他在 add 时先加锁，然后复制一份内容出来，在复制的内容上进行添加操作，然后再把数据引用指向复制的内容上。

public class CopyOnWriteArrayList<E>
    implements List<E>, RandomAccess, Cloneable, java.io.Serializable {
    final transient ReentrantLock lock = new ReentrantLock();
    private transient volatile Object[] array;
    final Object[] getArray() {
        return array;
    }
    final void setArray(Object[] a) {
        array = a;
    }
    
    // 省略大部分代码 ...  
    public boolean add(E e) {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            Object[] elements = getArray();
            int len = elements.length;
            Object[] newElements = Arrays.copyOf(elements, len + 1);
            newElements[len] = e;
            setArray(newElements);
            return true;
        } finally {
            lock.unlock();
        }
    }

    public E get(int index) {
        return get(getArray(), index);
    }
    // 省略大部分代码 ...
}

三、Map 接口线程安全实现类

3.1 普通 Map 变线程安全 Map

要想使普通 Map 能在多线程环境中使用，最简单的办法就是使用 Collections.synchronizedMap 方法，将普通 Map 转换为线程安全Map 。当然也可以手动加锁，如 synchronized。

如 HashMap 本身不是线程安全的，要使他具有线程安全可以使用 Collections.synchronizedMap(new HashMap<>()) 。然后使用时也使用普通 Map 并不差别。

Map<Object, Object> map = Collections.synchronizedMap(new HashMap<>());
// 添加键值对
map.put("aaa", 1);
map.put("bbb", 2);
map.put("ccc", 3);

SortedMap<String, String> sortedMap = new TreeMap<>();
SortedMap<String, String> synchronizedSortedMap = Collections.synchronizedSortedMap(sortedMap);

Google Guava 库中提供了线程安全的 Map 实现的 CopyOnWriteMap。

3.2 Hashtable

Hashtable 是一个线程安全的，他类似于 Vector 一样，在所有方法上加锁。

// 创建一个 Hashtable
Map<String, Integer> hashtable = new Hashtable<>();
// 添加键值对
hashtable.put("aaa", 1);
hashtable.put("bbb", 2);
hashtable.put("ccc", 3);

这是早期的产品，全部方法自带锁（synchronized），现在一般不用，要用一般也是将 HashMap 转成线程安全 Map。

3.3 ConcurrentHashMap

ConcurrentHashMap 是多线程环境下，效率最高的 Map。

//创建一个 ConcurrentHashMap
ConcurrentHashMap map = new ConcurrentHashMap ();
//添加元素
map.put("aaa", "111");
map.put("bbb", "aaa");
map.put("ccc", "aaa");

3.4 ConcurrentSkipListMap 有序/跳表

ConcurrentSkipListMap 是支持排序的并发集合，基于跳表（Skip List）数据结构实现。它能够在多线程并发环境中提供高效的有序映射。跳表的结构允许快速查找、插入和删除操作，同时保持有序性。

基于 HashMap 数据结构实现的 ConcurrentHashMap 表现在读效率高。TreeMap 基于红黑树实现的，能够排序。

在Redis 的 sorted_set 实现算法也是跳表
```
// 创建一个默认按照自然顺序排序的 ConcurrentSkipListMap
ConcurrentSkipListMap<Integer, String> skipListMap = new ConcurrentSkipListMap<>();
// 添加键值对
skipListMap.put(3, "Three");
skipListMap.put(1, "One");
skipListMap.put(2, "Two");
```

四、Set 接口线程安全实现类

4.1 普通 Set 变线程安全 Set

要想使普通 Set 能在多线程环境中使用，最简单的办法就是使用 Collections.synchronizedSet 方法，将普通 Set 转换为线程安全Set。当然也可以手动加锁，如 synchronized。
如 HashSet 本身不是线程安全的，要使他具有线程安全，可以使用 Collections.synchronizedSet(new HashSet<>()) 。接下来的方法与使用普通 Set 并不区别。
```
Set<Object> set = Collections.synchronizedSet(new HashSet<>());
set.add("aaa");
```

4.2 CopyOnWriteArraySet

CopyOnWriteArraySet 是 java.util.concurrent 包中提供的线程安全的 Set 实现。它使用 CopyOnWriteArrayList 来支持并发访问，特别适合读多写少的场景。
```
Set<String> set = new CopyOnWriteArraySet<>();
set.add("aaa");
```

4.3 ConcurrentSkipListSet 有序/跳表

ConcurrentSkipListSet 是 java.util.concurrent 包中基于跳表的实现，提供了有序的、线程安全的 Set 。与 ConcurrentSkipListMap 类似，它基于跳表数据结构，支持并发访问，并具有良好的性能特性。

Set<String> set = new ConcurrentSkipListSet<>();
set.add("aaa");

五、Queue 接口线程安全实现类

5.1 ConcurrentLinkedQueue 非阻塞/链表

ConcurrentLinkedQueue 是 java.util.concurrent 包中提供的非阻塞线程安全队列实现。它基于链表结构，适用于高并发的场景。

ConcurrentLinkedQueue 主要方法说明

peek() 获取队列中的值，但不从队列中删除该值。
poll() 获取队列中的值，且从队列中删除该值。
offer() 等同于 add 方法，都是向队列中添加数据。

ConcurrentLinkedQueue<String> concurrentLinkedQueue = new ConcurrentLinkedQueue<>();
// 添加元素
concurrentLinkedQueue.offer("aaa");
concurrentLinkedQueue.offer("bbb");
concurrentLinkedQueue.offer("ccc");
// 获取并移除队列头部元素
String headElement = concurrentLinkedQueue.poll();
// 获取队列头部元素（不移除）
String peekedElement = concurrentLinkedQueue.peek();
// 遍历元素
for (String item : concurrentLinkedQueue) {
    System.out.println(item);
}
// 检查元素是否存在
boolean containsBanana = concurrentLinkedQueue.contains("Banana");
// 获取队列大小
int size = concurrentLinkedQueue.size();
// 清空队列
concurrentLinkedQueue.clear();
// 检查是否为空
boolean isEmpty = concurrentLinkedQueue.isEmpty();

5.2 LinkedBlockingQueue 阻塞/链表

LinkedBlockingQueue 提供阻塞线程安全队列实现。它基于链表结构，可以在队列为空或已满时进行阻塞。

LinkedBlockingQueue<String> linkedBlockingQueue = new LinkedBlockingQueue<>();
// 添加元素
linkedBlockingQueue.offer("aaa");
linkedBlockingQueue.offer("bbb");
linkedBlockingQueue.offer("ccc");
// 获取并移除队列头部元素
try {
    String headElement = linkedBlockingQueue.take();
} catch (InterruptedException e) {
    e.printStackTrace();
}

5.3 ArrayBlockingQueue 阻塞/数组

ArrayBlockingQueue 提供阻塞线程安全队列实现。它基于数组结构，也可以在队列为空或已满时进行阻塞。

ArrayBlockingQueue 指定了长度的队列，基于数组，效率更高。

put() 方法向队列中添加元素，如果队列满了，则阻塞添加操作。
take() 方法从队列中获取元素，如果队列空了，则阻塞获取操作。

通过这两个方法可以实现了生产者、消费者模型。

package top.yiqifu.study.p002_collections;

import java.util.concurrent.ArrayBlockingQueue;

public class Test153_ArrayBlockingQueue {
    public static void main(String[] args) {
        // 创建一个 ArrayBlockingQueue，指定容量为3
        int capacity = 3;
        ArrayBlockingQueue<String> arrayBlockingQueue = new ArrayBlockingQueue<>(capacity);


        // 消费（获取）数据线程
        Thread thread1 = new Thread(() -> {
            for (int i = 1; i <= 100; i++) {

                try {
                    // 获取并移除队列头部元素
                    String headElement = arrayBlockingQueue.take();
                    System.out.println("获取数据 " + headElement);

                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        thread1.start();

        // 生产（添加）数据线程
        Thread thread2 = new Thread(() -> {
            for (int i = 1; i <= 100; i++) {

                try {
                    // 尝试添加元素（如果队列已满，会被阻塞）
                    arrayBlockingQueue.put("data " + i);
                    System.out.println("添加数据 " + i);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        thread2.start();


        try {
            thread1.join();
            thread2.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }
}

5.4 PriorityBlockingQueue 阻塞/数组

PriorityBlockingQueue 阻塞线程安全队列实现，具有优先级排序。元素按照它们的自然顺序或根据构造函数提供的比较器进行排序。

package top.yiqifu.study.p002_collections;

import java.util.concurrent.*;

public class Test154_PriorityBlockingQueue {
    public static void main(String[] args) {
        PriorityBlockingQueue<String> priorityBlockingQueue = new PriorityBlockingQueue<>();
        // 添加元素
        priorityBlockingQueue.offer("bbb");
        priorityBlockingQueue.offer("ccc");
        priorityBlockingQueue.offer("aaa");

        // 遍历元素，遍历不一定有序，遍历不移除
        for (String item : priorityBlockingQueue) {
            System.out.println(item);
        }

        // 获取并移除队列头部元素（take 按照自然排序顺序）
        while (priorityBlockingQueue.size()>0) {
            try {
                String headElement = priorityBlockingQueue.take();
                System.out.println("take=" + headElement);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

    }
}

5.5 DelayQueue 阻塞/时间排序

DelayQueue 提供阻塞线程安全队列实现，用于按照指定的延迟时间对元素进行排序。即按等待时间排序，使用 PriorityQueue 实现。

PriorityQueue 是具有排序功能（使用二叉树排序）的队列

DelayQueue 中，使用 take 方法时，按等待时间短的先返回。即可以按时间进行任务调度。

package top.yiqifu.study.p002_collections;


import java.util.concurrent.BlockingQueue;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;

public class Test155_DelayQueue {
    public static void main(String[] args) throws InterruptedException {
        // 创建一个 DelayQueue
        BlockingQueue<DelayedElement> delayQueue = new DelayQueue<>();

        // 添加延迟元素
        delayQueue.put(new DelayedElement("Element1", 3000));
        delayQueue.put(new DelayedElement("Element2", 2000));
        delayQueue.put(new DelayedElement("Element3", 5000));

        // 遍历元素
        System.out.println("遍历元素:");
        for (DelayedElement element : delayQueue) {
            System.out.println(element);
        }
        System.out.println();

        // 获取队列头部元素（不移除）
        DelayedElement peekedElement = delayQueue.peek();
        System.out.println("获取队列头部元素（不移除）: " + peekedElement);
        System.out.println();

        while (delayQueue.size()>0) {
            // 获取并移除队列头部元素（按照剩余延迟时间排序）
            try {
                DelayedElement headElement = delayQueue.take(); // 注意这个方法在时间没到时会阻塞
                System.out.println("Taked Head Element: " + headElement);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

    }



    static class DelayedElement implements Delayed {
        private String name;
        private long delayTime;

        public DelayedElement(String name, long delayTime) {
            this.name = name;
            this.delayTime = System.currentTimeMillis() + delayTime;
        }

        @Override
        public long getDelay(TimeUnit unit) {
            long diff = delayTime - System.currentTimeMillis();
            return unit.convert(diff, TimeUnit.MILLISECONDS);
        }

        @Override
        public int compareTo(Delayed o) {
            return Long.compare(this.delayTime, ((DelayedElement) o).delayTime);
        }

        @Override
        public String toString() {
            return "DelayedElement{" +
                    "name='" + name + '\'' +
                    ", delayTime=" + delayTime +
                    '}';
        }
    }
}

5.6 SynchronousQueue 一对一同步

SynchronousQueue 是一种特殊的队列，它的大小始终为 0。由于其特殊性，SynchronousQueue 并不会存储元素，而是在生产者和消费者之间进行直接传递。其安全性是通过直接传递而非存储实现的。

没有容量的队列

不能 add ，只能 put

作用：把一个数据传递到另一个线程。线程间数据交换。有点类似于 Exchanger。

package top.yiqifu.study.p002_collections;


import java.util.concurrent.SynchronousQueue;

public class Test156_SynchronousQueue {
    public static void main(String[] args)  {
        // 创建一个 SynchronousQueue
        SynchronousQueue<String> synchronousQueue = new SynchronousQueue<>();

        // 启动消费者线程
        new Thread(() -> {
            for (int i = 1; i < 10; i++) {
                try {
                    System.out.println("准备消费"+i);
                    String message = synchronousQueue.take(); // 阻塞直到有生产者放入元素
                    System.out.println("消费: " + message);
                    System.out.println("消费完成"+i);

                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();

        // 启动生产者线程
        new Thread(() -> {
            for (int i = 1; i < 10; i++) {
                try {
                    String message = "同步消息!"+i;
                    System.out.println("生产: " + message);
                    synchronousQueue.put(message); // 阻塞直到有消费者接收
                    // synchronousQueue.put(message+"-new"); // 等前面阻塞了这句才会执行，即同一时刻队列中只有一个。
                    System.out.println("生产完成"+i);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();



    }

}

5.7 LinkedTransferQueue 多对多同步

LinkedTransferQueue 支持高并发的生产者-消费者模型。提供了 TransferQueue 接口的实现，允许生产者等待消费者来消费元素。

LinkedTransferQueue 的 transfer 方法不管满不满都先阻塞，与 SynchronousQueue 不同的，他可以在多个生产者线程中使用，实现多个生产者等一个消费者。或者实现多个生产者等多个消费者。

package top.yiqifu.study.p002_collections;


import java.util.concurrent.LinkedTransferQueue;

public class Test157_LinkedTransferQueue {
    public static void main(String[] args) {
        // 创建一个 LinkedTransferQueue
        LinkedTransferQueue<String> transferQueue = new LinkedTransferQueue<>();

        // 启动生产者线程
        for (int t = 1; t <= 3 ; t++) {
            new Thread(() -> {
                for (int i = 1; i <= 2; i++) {
                    try {
                        String producor = Thread.currentThread().getName();
                        String message = "同步消息!" + i;
                        System.out.println(producor+"生产: " + message);
//                        transferQueue.put(message + " by put 1"); // 不阻塞
//                        transferQueue.put(message + " by put 2");// 不阻塞
//                        System.out.println(producor+"生产put完成");
                        transferQueue.transfer(message + " by transfer");// 阻塞直到有消费者接收
                        System.out.println(producor+"生产transfer完成" + i);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }, "生产者"+t).start();
        }

        // 启动消费者线程
        new Thread(() -> {
            for (int i = 1; i < 20; i++) {
                try {
                    System.out.println("消费准备"+i);
                    String message = transferQueue.take(); // 阻塞直到有生产者放入元素
                    System.out.println("消费: " + message);
                    System.out.println("消费完成"+i);

                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }

}

文章来源:https://blog.csdn.net/qifu123/article/details/134977425
本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若内容造成侵权/违法违规/事实不符，请联系我的编程经验分享网邮箱：veading@qq.com进行投诉反馈，一经查实，立即删除！