java

JUC并发编程(六)——并发容器

Nick · 7月6日 · 2025年 · 本文7130字 · 阅读18分钟204

并发容器概览

  • ConcurrentHashMap：线程安全的HashMap
  • CopyOnWriteArrayList：线程安全的List
  • BlockingQueue：这是一个接口，表示阻塞队列，非常适合用于作为数据共享的通道
  • ConcurrentLinkedQueue: 高效的非阻塞并发队列，使用
    链表实现。可以看做一个线程安全的LinkedList
  • ConcurrentSkipListMap:是一个Map，使用跳表的数据结
    构进行快速查找

集合类的历史

  1. Vector和Hashtable
    线程安全的

JUC并发编程(六)——并发容器-Nickの博客

大部分方法都是使用synchronized修饰，性能较差。

  1. ArrayList和HashMap
    非线程安全的

ArrayList线程安全解决方案：

  • 手动加锁（性能开销大）
  • 使用Collections.synchronizedList(new ArrayList<>)（性能开销较大）
  • 使用CopyOnWriteArrayList（写时拷贝，适用于读多写少场景）

HashMap线程安全解决方案：

  • 使用Hashtable（性能较差，不推荐）
  • 使用ConcurrentHashMap（读无锁、写分段锁，推荐）
  1. ConcurrentHashMap和CopyOnWriteArrayList
    取代同步的HashMap和同步的ArrayList，绝大多数并发情况下，ConcurrentHashMap和
    CopyOnWriteArrayList的性能都更好

为什么HashMap是线程不安全的？
同时put碰撞导致数据丢失
同时put扩容导致数据丢失
在扩容的时候会造成链表的死循环，CPU100%

参考博客：Java HashMap的死循环

HashMap数据结构

JDK1.8后HashMap采用“数组 + 链表 + 红黑树”复合结构，通过哈希表实现高效存储与检索数据结构，初始容量16、负载因子0.75平衡内存与性能。

数组（桶数组）：

  • 作为底层存储结构，每个元素称为“桶”（Bucket）。
  • 每个桶存储一个链表头节点或红黑树根节点。
  • 数组长度为2的幂次方（如16、32），通过位运算优化哈希索引。

链表：

  • 解决哈希冲突（不同键值对映射到同一桶），形成链表。
  • 每个节点（Node）包含键值对和指向下一个节点的引用。

红黑树（JDK 8+）：

  • 当链表长度超过阈值（默认8）且数组长度≥64时，链表转红黑树。
  • 提高查找效率，避免极端情况下的性能退化（如哈希冲突导致链表过长）。

红黑树简介：
对二叉查找树BST的一种平衡策略， 红黑树查找时间复杂度为 O(log n)，链表为 O(n)。
会自动平衡，防止极端不平衡从而影响查找效率的情况发生

ConcurrentHashMap

ConcurrentHashMap实现和分析

JDK1.7的ConcurrentHashMap实现和分析

  • Java 7中的ConcurrentHashMap最外层是多个segment，每个segment的底层数据结构与HashMap类似，仍然是数组和链表组成的拉链法
  • 每个segment独立上ReentrantLock锁，每个segment之间互不影响，提高了并发效率
  • ConcurrentHashMap 默认有16个 Segments，所以最多可以同时支持16 个线程并发写（操作分别分布在不同的
    Segment 上）。这个默认值可以在初始化的时候设置为其他值，但是一旦初始化以后，是不可以扩容的

JUC并发编程(六)——并发容器-Nickの博客

JDK1.8的ConcurrentHashMap实现和分析

  • Java 8中的ConcurrentHashMap最外层是多个Node，链表长度≥8且数组长度≥64时自动转红黑树
  • 头节点synchronized锁（细粒度锁），通过CAS和volatile保证原子性，单个数组支持更高并发
  • 1.8采用细粒度锁（头节点synchronized），可扩容，显著提升高并发场景下的吞吐量

JUC并发编程(六)——并发容器-Nickの博客

putVal流程

  1. 判断key value 不为空
  2. 计算hash值
  3. 根据对应位置节点的类型，来赋值，或者 helpTransfer，或者增长链表，或者给红黑树增加节点
  4. 检查满足阈值就“红黑树化”
  5. 返回oldVal

注意：链表转红黑树阈值：8

原因：基于泊松分布计算，当负载因子为0.75时，链表长度达到8的概率极低（约0.00000006），属于小概率事件。
设计目的：在正常哈希分布下优先使用高效链表结构，仅在极端冲突时触发红黑树优化。

get流程

  1. 计算hash值
  2. 找到对应的位置，根据情况进行：
  3. 直接取值
  4. 红黑树里找值
  5. 遍历链表取值
  6. 返回找到的结果

为什么要把1.7的结构改成1.8的结构？
JUC并发编程(六)——并发容器-Nickの博客

ConcurrentHashMap线程安全注意点

组合操作并不保证线程安全

案例代码：

package collections.concurrenthashmap;

import java.util.concurrent.ConcurrentHashMap;

/**
 * 描述：     组合操作并不保证线程安全
 */
public class OptionsNotSafe implements Runnable {

    private static ConcurrentHashMap<String, Integer> scores = new ConcurrentHashMap<String, Integer>();

    public static void main(String[] args) throws InterruptedException {
        scores.put("小明", 0);
        Thread t1 = new Thread(new OptionsNotSafe());
        Thread t2 = new Thread(new OptionsNotSafe());
        t1.start();
        t2.start();
        t1.join();
        t2.join();
        System.out.println(scores);
    }

    @Override
    public void run() {
        for (int i = 0; i < 1000; i++) {
            Integer score = scores.get("小明");
            Integer newScore = score + 1;
            scores.put("小明",newScore);
        }

    }
}

运行结果
JUC并发编程(六)——并发容器-Nickの博客
发现组合操作并非线程安全，concurrenthashmap只能保证get或者put等操作的线程安全。

代码优化：

package collections.concurrenthashmap;

import java.util.concurrent.ConcurrentHashMap;

/**
 * 描述：     组合操作并不保证线程安全
 */
public class OptionsNotSafe implements Runnable {

    private static ConcurrentHashMap<String, Integer> scores = new ConcurrentHashMap<String, Integer>();

    public static void main(String[] args) throws InterruptedException {
        scores.put("小明", 0);
        Thread t1 = new Thread(new OptionsNotSafe());
        Thread t2 = new Thread(new OptionsNotSafe());
        t1.start();
        t2.start();
        t1.join();
        t2.join();
        System.out.println(scores);
    }

    @Override
    public void run() {
        for (int i = 0; i < 1000; i++) {
            while (true) {
                Integer score = scores.get("小明");
                Integer newScore = score + 1;
                boolean b = scores.replace("小明", score, newScore);
                if (b) {
                    break;
                }
            }
        }

    }
}

运行结果：
JUC并发编程(六)——并发容器-Nickの博客

使用scores.replace加自旋的方式保证了组合操作的线程安全。

类似的组合操作的常用方法还有putlfAbsent，等价于下面代码。

if (!map.containsKey(key))
    return map.put(key, value)；
else
    return map.get(key);
}

CopyOnWriteArrayList

诞生原因：
Vector和SynchronizedList的锁的粒度太大，并发效率相对比较低，并且迭代时无法编辑
Copy-On-Write并发容器还包括CopyOnWriteArraySet，用
来替代同步Set

简介

CopyOnWriteArrayList 是 Java 集合框架中一个线程安全的 List 实现类，它是 ArrayList 的并发版本，适用于读多写少的场景。其核心思想是写时复制（Copy-On-Write, 简称 COW），即在进行写操作（如添加、修改元素）时，并不会直接修改原始数组，而是先复制一份当前数组的副本，在副本上进行修改操作，完成后再将引用指向新的数组，从而保证读操作的线程安全性。

这种机制使得读操作可以无锁执行，提高了并发读取的性能。由于写操作涉及数组复制，因此在写操作频繁的场景下，性能会受到影响，同时也会带来额外的内存开销。

适用场景

适用场景：

  1. 读操作可以尽可能地快，而写即使慢一些也没有太大关系
  2. 读多写少：黑名单，每日更新；监听器：迭代操作远多余修改操作

读写规则：

  1. 回顾读写锁：读读共享、其他都互斥（写写互斥、读写互斥、写读互斥）
  2. 读写锁规则的升级：读取是完全不用加锁的，并且更厉害的是，写入也不会阻塞读取操作。只有写入和写入之间需要进行同步等待
  3. 读取迭代的时候，会先创建副本，利用副本进行迭代，但是在迭代过程中对数据进行操作，操作的是原数据，和创建的副本没有关系，所以迭代无感知。

写入操作源码：
JUC并发编程(六)——并发容器-Nickの博客

案例分析：

package collections.copyonwrite;

import java.util.Iterator;
import java.util.concurrent.CopyOnWriteArrayList;

/**
 * 描述：     对比两个迭代器
 */
public class CopyOnWriteArrayListDemo2 {

    public static void main(String[] args) throws InterruptedException {
        CopyOnWriteArrayList<Integer> list = new CopyOnWriteArrayList<>(new Integer[]{1, 2, 3});

        System.out.println(list);

        Iterator<Integer> itr1 = list.iterator();

        list.remove(2);
        Thread.sleep(1000);
        System.out.println(list);

        Iterator<Integer> itr2 = list.iterator();

        itr1.forEachRemaining(System.out::println);
        itr2.forEachRemaining(System.out::println);

    }
}

运行结果：
JUC并发编程(六)——并发容器-Nickの博客

迭代器1创建后，修改数据，并不会影响迭代器1的数据内容，迭代器2后创建的，则是修改后的内容。

CopyOnWriteArrayList的缺点

  1. 数据一致性问题：CopyOnWrite容器只能保证数据的最终一致性，不能保证数据的实时一致性。所以如果你希望写入的数据，马上能读到，请不要使用CopyOnWrite容器。
  2. 内存占用问题：因为CopyOnWrite的写是复制机制，所以在进行写操作的时候，内存里会同时驻扎两个对象的内存。

并发队列Queue：阻塞队列

各并发队列关系图：

JUC并发编程(六)——并发容器-Nickの博客

BlockingQueue阻塞队列简介

什么是阻基队列？
阻塞队列是具有阻塞功能的队列，所以它首先是一个队列，其次是具有阻塞功能。
通常，阻塞队列的一端是给生产者放数据用，另一端给消费者拿数据用。阻塞队列是线程安全的，所以生产者和消费者都可以是多线程的

阻塞功能：两个带有阻塞功能的常用方法是：
take()方法：获取并移除队列的头结点，一旦如果执行take的时候，队列里无数据，则阻塞，直到队列里有数据

put()方法：插入元素。但是如果队列己满，那么就无法继续插入，则阻塞，直到队列里有了空闲空间

add, remove, element（返回头节点）：特点是如果队列满了，对列为空，则会抛出异常

offer(添加元素), poll(取出一个元素并删除), peek(取出元素不删除)：特点是返回布尔值，不会抛异常

ArrayBlockingQueue

简介和案例

特点：

  • 有界
  • 指定容量
  • 公平：还可以指定是否需要保证公平，如果想保证公平的话那么等待了最长时间的线程会被优先处理，不过这会同时带来一定的性能损耗

使用案例
有10个面试者，一共只有1个面试官，大厅里有3个位子供面试者休息，每个人的面试时间是10秒广档州所有人面试的场景

案例代码：

package collections.queue;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

/**
 * 描述：     TODO
 */
public class ArrayBlockingQueueDemo {

    public static void main(String[] args) {

        ArrayBlockingQueue<String> queue = new ArrayBlockingQueue<String>(3);

        Interviewer r1 = new Interviewer(queue);
        Consumer r2 = new Consumer(queue);
        new Thread(r1).start();
        new Thread(r2).start();
    }
}

class Interviewer implements Runnable {

    BlockingQueue<String> queue;

    public Interviewer(BlockingQueue queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        System.out.println("10个候选人都来啦");
        for (int i = 0; i < 10; i++) {
            String candidate = "Candidate" + i;
            try {
                queue.put(candidate);
                System.out.println("安排好了" + candidate);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        try {
            queue.put("stop");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

class Consumer implements Runnable {

    BlockingQueue<String> queue;

    public Consumer(BlockingQueue queue) {

        this.queue = queue;
    }

    @Override
    public void run() {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        String msg;
        try {
            while(!(msg = queue.take()).equals("stop")){
                System.out.println(msg + "到了");
            }
            System.out.println("所有候选人都结束了");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

运行结果：
JUC并发编程(六)——并发容器-Nickの博客

主要在多线程中，使用了BlockingQueue，容量为3的阻塞队列，put方法添加元素和take方法取元素实现有序的生产者和消费者模式。

Linked BlockingQueue

  • 无界
  • 容量Integer.MAX_VALUE
  • 内部结构：Node、两把锁

Priority Blocking Queue

  • 支持优先级
  • 自然顺序（而不是先进先出）
  • 无界队列
  • PriorityQueue 的线程安全版本

SynchronousQueue

  • 它的容量为0
  • 需要注意的是，SynchronousQueue的容量不是1而是0，因为SynchronousQueue不需要去持有元素，它所做的就是直接传递 ( direct handoff ）
  • 效率很高
  • SynchronousQueue没有peek等函数，因为peek的含义是取出头结点，但是SynchronousQueue的容量是0，所以连头结点都没有，也就没有peek方法。同理，没有iterate相关方法
  • 是一个极好的用来直接传递的并发数据结构
  • SynchronousQueue是线程池Executors.newCachedThreadPool()使用的阻塞队列

DelayQueue

  • 延迟队列，根据延迟时间排序
  • 元素需要实现Delayed接口，规定排序规则

非阻塞并发队列

并发包中的非阻塞队列只有ConcurrentLinkedQueue这一种顾名思义ConcurrentLinkedQueue是使用链表作为其数据结构的，使用 CAS 非阻塞算法来实现线程安全（不具备阻塞功能），适合用在对性能要求较高的并发场景。用的相对比较少一些
看源码的offer方法的CAS思想，内有p.casNext方法，用了UNSAFE.compareAndSwapObject

JUC并发编程(六)——并发容器-Nickの博客

相关文章

JUC并发编程(九)——Future和Callable(获取子线程结果)

9月13日 · 2025年

JUC并发编程(八)——AQS(并发灵魂)

8月6日 · 2025年

JUC并发编程(七)——控制并发流程

7月27日 · 2025年

JUC并发编程(五)——CAS原理&final不变性

6月15日 · 2025年
0 条回应
©2021版权所有
在线人数：1人 来访统计
新时代
Ado
隐藏