并发集合与阻塞队列

学习目标

选择 ConcurrentHashMap、CopyOnWriteArrayList 和 BlockingQueue 的合适场景。
理解弱一致迭代、快照迭代、原子复合方法和容量背压。
知道非阻塞队列、跳表集合和阻塞队列的差异。

核心知识点

专业术语

ConcurrentHashMap：支持并发读写的 Map，单次方法和指定复合操作有并发语义。
CopyOnWriteArrayList：写时复制列表，迭代器读取稳定快照，写成本高。
BlockingQueue：队列满/空时可阻塞等待，天然表达生产者—消费者背压。
弱一致迭代（weakly consistent）：遍历不抛 ConcurrentModificationException，可能看到迭代开始前后的一部分更新。
快照迭代：迭代器固定在创建时的数组，不反映之后的写入。

白话解释与边界

并发集合只解决容器内部的并发结构，不会替你定义业务事务。ConcurrentHashMap 的 putIfAbsent、compute 等方法可以表达特定原子边界；多次普通 get/put 仍可能竞态。CopyOnWriteArrayList 适合读多写少、列表较小的配置或监听器，不适合频繁写入的大数据。

BlockingQueue 通过容量限制把生产速度和消费速度连接起来。put/take 会等待，offer/poll 可立即返回或带超时；无界队列并不等于没有风险，堆积最终仍会消耗内存。

常用用法

ConcurrentHashMap.putIfAbsent：只初始化一次

putIfAbsent 返回旧值；返回 null 表示这次写入了新值。

java

// 说明：ConcurrentHashMap.putIfAbsent：只初始化一次。
import java.util.concurrent.ConcurrentHashMap;

public class ConcurrentMapPutIfAbsentDemo {
    public static void main(String[] args) {
        var map = new ConcurrentHashMap<String, Integer>();
// 作用：var map = new ConcurrentHashMap<String, Integer>();；只初始化一次。
// 初始状态：map = new ConcurrentHashMap<String, Integer>()。
        Integer first = map.putIfAbsent("java", 20);
// 初始状态：first = map.putIfAbsent("java", 20)。
// 作用：Integer first = map.putIfAbsent("java", 20);；只初始化一次。
        Integer second = map.putIfAbsent("java", 21);
// 初始状态：second = map.putIfAbsent("java", 21)。
// 作用：Integer second = map.putIfAbsent("java", 21);；只初始化一次。
        System.out.println(first + ", " + second + ", value=" + map.get("java"));
// 输出：null, 20, value=20
    }
}

需要根据旧值做计算时使用 compute 或 merge，避免把 get 和 put 拆成两个竞态步骤。

ConcurrentHashMap.computeIfAbsent：并发缓存

计算函数应短小、无副作用，不能递归修改同一个 Map，也不应在 Map 的计算锁内执行长时间 I/O。

java

// 说明：ConcurrentHashMap.computeIfAbsent：并发缓存。
import java.util.concurrent.ConcurrentHashMap;

public class ConcurrentMapCacheDemo {
    public static void main(String[] args) {
        var cache = new ConcurrentHashMap<String, String>();
// 作用：var cache = new ConcurrentHashMap<String, String>();；并发缓存。
// 初始状态：cache = new ConcurrentHashMap<String, String>()。
        String value = cache.computeIfAbsent("java", String::toUpperCase);
// 初始状态：value = cache.computeIfAbsent("java", String::toUpperCase)。
// 作用：String value = cache.computeIfAbsent("java", String::toUpperCase);；并发缓存。
        System.out.println(value + ", size=" + cache.size());
// 输出：JAVA, size=1
    }
}

ConcurrentHashMap.compute：按旧值原子更新

compute 把读、计算和写入放进 Map 定义的复合操作中；函数返回 null 会删除已有映射，是否允许这种语义要写清楚。

java

// 说明：ConcurrentHashMap.compute：按旧值原子更新。
import java.util.concurrent.ConcurrentHashMap;

public class ConcurrentMapComputeDemo {
    public static void main(String[] args) {
        var counts = new ConcurrentHashMap<String, Integer>();
// 作用：var counts = new ConcurrentHashMap<String, Integer>();；按旧值原子更新。
// 初始状态：counts = new ConcurrentHashMap<String, Integer>()。
        counts.put("java", 1);
// 作用：counts.put("java", 1);；按旧值原子更新。
        counts.compute("java", (key, old) -> old == null ? 1 : old + 1);
// 初始状态：counts.compute("java", (key, old) -> old == null ? 1 : old + 1)。
// 作用：counts.compute("java", (key, old) -> old == null ? 1 : old + 1);；按旧值原子更新。
        System.out.println(counts.get("java"));
// 输出：2
    }
}

ConcurrentHashMap.merge：累计或合并值

merge 适合计数、合并列表等按键聚合；合并函数应保持轻量且可重复计算，不能依赖外部可变状态。

java

// 说明：ConcurrentHashMap.merge：累计或合并值。
import java.util.concurrent.ConcurrentHashMap;

public class ConcurrentMapMergeDemo {
    public static void main(String[] args) {
        var counts = new ConcurrentHashMap<String, Integer>();
// 作用：var counts = new ConcurrentHashMap<String, Integer>();；累计或合并值。
// 初始状态：counts = new ConcurrentHashMap<String, Integer>()。
        counts.merge("java", 1, Integer::sum);
// 初始状态：counts.merge("java", 1, Integer::sum)。
// 作用：counts.merge("java", 1, Integer::sum);；累计或合并值。
        counts.merge("java", 2, Integer::sum);
// 初始状态：counts.merge("java", 2, Integer::sum)。
// 作用：counts.merge("java", 2, Integer::sum);；累计或合并值。
        System.out.println(counts.get("java"));
// 输出：3
    }
}

CopyOnWriteArrayList：读快照

迭代器基于创建时数组，不会看到之后追加的 metrics；每次写都会复制数组，因此监听器数量少、读远多于写时才合适。

java

// 说明：CopyOnWriteArrayList：读快照。
import java.util.concurrent.CopyOnWriteArrayList;

public class CopyOnWriteDemo {
    public static void main(String[] args) {
        var listeners = new CopyOnWriteArrayList<String>();
// 初始状态：listeners = new CopyOnWriteArrayList<String>()。
// 作用：var listeners = new CopyOnWriteArrayList<String>();；读快照。
        listeners.add("audit");
// 作用：listeners.add("audit");；读快照。
        var snapshot = listeners.iterator();
// 作用：var snapshot = listeners.iterator();；读快照。
// 初始状态：snapshot = listeners.iterator()。
        listeners.add("metrics");
// 作用：listeners.add("metrics");；读快照。
        while (snapshot.hasNext()) {
            System.out.println(snapshot.next());
// 输出：audit
        }
        System.out.println(listeners);
// 输出：[audit, metrics]
    }
}

阻塞式交接：放入并取出队列元素

有界队列满时 put 等待，空时 take 等待；生产者和消费者都应设置退出、中断和超时策略，避免关闭时永久阻塞。

java

// 说明：阻塞式交接：放入并取出队列元素。
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public class ArrayBlockingQueueDemo {
    public static void main(String[] args) throws InterruptedException {
        BlockingQueue<String> queue = new ArrayBlockingQueue<>(1);
// 初始状态：queue = new ArrayBlockingQueue<>(1)。
// 作用：BlockingQueue<String> queue = new ArrayBlockingQueue<>(1);；放入并取出队列元素。
        queue.put("job-1");
// 初始状态：queue.put("job-1")。
// 作用：queue.put("job-1");；放入并取出队列元素。
        System.out.println(queue.take());
// 输出：job-1
    }
}

有界队列交互：立即尝试放入或取出

offer/poll 立即返回，带超时重载可以在有限时间内等待。

java

// 说明：有界队列交互：立即尝试放入或取出。
import java.util.concurrent.ArrayBlockingQueue;

public class OfferPollDemo {
    public static void main(String[] args) {
        var queue = new ArrayBlockingQueue<String>(1);
// 初始状态：queue = new ArrayBlockingQueue<String>(1)。
// 作用：var queue = new ArrayBlockingQueue<String>(1);；立即尝试放入或取出。
        boolean accepted = queue.offer("job");
// 初始状态：accepted = queue.offer("job")。
// 作用：boolean accepted = queue.offer("job");；立即尝试放入或取出。
        String value = queue.poll();
// 初始状态：value = queue.poll()。
// 作用：String value = queue.poll();；立即尝试放入或取出，返回读取结果。
        System.out.println(accepted + ", " + value);
// 输出：true, job
    }
}

返回 false 或 null 时要明确是丢弃、重试、降级还是关闭，不要直接忽略。

LinkedBlockingQueue：可选容量的链式阻塞队列

显式指定容量才能形成内存边界；不带容量的构造器上限很大，仍应通过监控和业务限制避免无限积压。

java

// 说明：LinkedBlockingQueue：可选容量的链式阻塞队列。
import java.util.concurrent.LinkedBlockingQueue;

public class LinkedBlockingQueueDemo {
    public static void main(String[] args) throws InterruptedException {
        var queue = new LinkedBlockingQueue<String>(2);
// 初始状态：queue = new LinkedBlockingQueue<String>(2)。
// 作用：var queue = new LinkedBlockingQueue<String>(2);；可选容量的链式阻塞队列。
        queue.put("a");
// 作用：queue.put("a");；可选容量的链式阻塞队列。
        queue.put("b");
// 作用：queue.put("b");；可选容量的链式阻塞队列。
        System.out.println(queue.size() + ", " + queue.take());
// 输出：2, a
    }
}

不常用但需要知道

ConcurrentLinkedQueue：无阻塞无容量队列

空队列立即返回 null，不提供等待或容量背压；适合轻量事件收集，不适合作为有界任务队列。

java

// 说明：queue 没有容量上限；offer("a") 不会等待空位，poll() 也不会等待新元素。
import java.util.concurrent.ConcurrentLinkedQueue;

public class ConcurrentLinkedQueueDemo {
    public static void main(String[] args) {
        var queue = new ConcurrentLinkedQueue<String>();
// 初始状态：queue = new ConcurrentLinkedQueue<String>()。
// 作用：var queue = new ConcurrentLinkedQueue<String>();；无阻塞无容量队列。
        queue.offer("a");
// 作用：queue.offer("a");；无阻塞无容量队列。
        System.out.println(queue.poll());
// 输出：a
    }
}

ConcurrentSkipListMap：并发有序 Map

它提供并发排序和范围视图，通常比 ConcurrentHashMap 成本高；只需要按键查找时优先 HashMap 家族。

java

// 说明：map 并发维持 Integer 键的自然顺序；先 put(2) 再 put(1) 之后 firstEntry() 仍返回 1=a。
import java.util.concurrent.ConcurrentSkipListMap;

public class SkipListMapDemo {
    public static void main(String[] args) {
        var map = new ConcurrentSkipListMap<Integer, String>();
// 初始状态：map = new ConcurrentSkipListMap<Integer, String>()。
// 作用：var map = new ConcurrentSkipListMap<Integer, String>();；并发有序 Map。
        map.put(2, "b");
// 作用：map.put(2, "b");；并发有序 Map。
        map.put(1, "a");
// 作用：map.put(1, "a");；并发有序 Map。
        System.out.println(map.firstEntry());
// 输出：1=a
    }
}

LinkedTransferQueue.transfer：等待消费者接收

transfer 会等待一个消费者接收元素，比普通 put 更强调“交接完成”；需要简单排队时 BlockingQueue 更直观。上例使用普通平台线程接收元素，不依赖虚拟线程预览 API。

java

// 作用：通过 LinkedTransferQueue.transfer 等待消费者接收。
import java.util.concurrent.LinkedTransferQueue;

public class TransferQueueDemo {
    public static void main(String[] args) throws InterruptedException {
        var queue = new LinkedTransferQueue<String>();
// 作用：var queue = new LinkedTransferQueue<String>();；等待消费者接收。
// 初始状态：queue = new LinkedTransferQueue<String>()。
        Thread consumer = new Thread(() -> {
// 作用：Thread consumer = new Thread(() -> {；等待消费者接收。
// 初始状态：consumer = new Thread(() ->。
            try {
                System.out.println(queue.take());
// 输出：direct
            } catch (InterruptedException ex) {
                Thread.currentThread().interrupt();
// 作用：Thread.currentThread().interrupt();；等待消费者接收。
            }
        }, "consumer");
        consumer.start();
// 作用：consumer.start();；等待消费者接收。
        queue.transfer("direct");
// 初始状态：queue.transfer("direct")。
// 作用：queue.transfer("direct");；等待消费者接收。
        consumer.join();
// 作用：consumer.join();；等待消费者接收。
    }
}

`keySet` 视图：并发 Map 的集合视图

视图与底层 Map 共享数据，适合并发去重集合；需要 Map 值或独立快照时不要误用它。

java

// 说明：newKeySet() 返回由 ConcurrentHashMap 支撑的可变 Set；keys.add("java") 是线程安全的单键插入。
import java.util.concurrent.ConcurrentHashMap;

public class ConcurrentKeySetDemo {
    public static void main(String[] args) {
        var keys = ConcurrentHashMap.<String>newKeySet();
// 初始状态：keys = ConcurrentHashMap.<String>newKeySet()。
        keys.add("java");
// 初始状态：keys.add("java")。
// 作用：keys.add("java");；并发 Map 的集合视图。
        System.out.println(keys.contains("java"));
// 输出：true
    }
}

简单案例

java

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

public class ProducerConsumerDemo {
    public static void main(String[] args) throws InterruptedException {
        BlockingQueue<String> queue = new ArrayBlockingQueue<>(2);
// 初始状态：queue = new ArrayBlockingQueue<>(2)。
// 作用：BlockingQueue<String> queue = new ArrayBlockingQueue<>(2);；并发 Map 的集合视图。
        queue.put("parse");
// 初始状态：queue.put("parse")。
// 作用：queue.put("parse");；并发 Map 的集合视图。
        queue.put("save");
// 初始状态：queue.put("save")。
// 作用：queue.put("save");；并发 Map 的集合视图。
        System.out.println(queue.take() + " -> " + queue.take());
// 输出：parse -> save
    }
}

队列让生产和消费通过消息边界解耦；真实多线程版本还要处理 poison pill、interrupt、超时和关闭顺序。

易混点

ConcurrentHashMap 的单次方法安全，不等于 get 后再 put 的复合业务安全。
CopyOnWriteArrayList 的迭代器是快照，写入会复制数组，写多时性能很差。
BlockingQueue 会等待并提供背压，ConcurrentLinkedQueue 立即返回且没有容量限制。
poll() 返回 null 既可能是空队列，也可能是业务允许的 null 值；BlockingQueue 通常不允许 null。
有序并发集合应选择 ConcurrentSkipListMap/Set，不要用同步包装器假装解决所有性能问题。

课后小问

为什么缓存初始化更适合 computeIfAbsent 而不是 containsKey + put？答案：computeIfAbsent 把检查和写入组合成 Map 的并发原子边界，避免多个线程重复初始化或覆盖。解析：计算函数仍应短小无副作用，复杂加载要配合超时、失败清理和缓存策略。
CopyOnWriteArrayList 为什么不适合高频写入？答案：每次结构性写入都会复制底层数组，写放大和内存分配成本很高。解析：它的收益来自读多写少和稳定迭代，写多场景应考虑锁、并发队列或其他结构。
队列容量为什么是并发系统的重要参数？答案：容量决定生产过快时是等待、拒绝、降级还是继续占用内存。解析：没有容量边界的积压会把问题从延迟拖到内存耗尽，容量和拒绝策略要配套监控。

本节小结

ConcurrentHashMap 用原子复合方法表达并发缓存和计数，普通多步读写仍需设计。
CopyOnWriteArrayList 以写成本换取读快照，适合读多写少。
BlockingQueue 通过阻塞和容量实现生产者—消费者背压。
ConcurrentLinkedQueue、ConcurrentSkipListMap、TransferQueue 是更低频的语义特化工具。

快速回顾

会用 putIfAbsent、computeIfAbsent、compute 和 merge。
能解释 CopyOnWriteArrayList 的快照与写时复制。
会在 put/take 与 offer/poll 之间选择等待或非等待策略。
能区分阻塞队列、无阻塞队列和并发有序 Map。

并发集合与阻塞队列 ​

学习目标 ​

核心知识点 ​

专业术语 ​

白话解释与边界 ​

常用用法 ​

ConcurrentHashMap.putIfAbsent：只初始化一次 ​

ConcurrentHashMap.computeIfAbsent：并发缓存 ​

ConcurrentHashMap.compute：按旧值原子更新 ​

ConcurrentHashMap.merge：累计或合并值 ​

CopyOnWriteArrayList：读快照 ​

阻塞式交接：放入并取出队列元素 ​

有界队列交互：立即尝试放入或取出 ​

LinkedBlockingQueue：可选容量的链式阻塞队列 ​

不常用但需要知道 ​

ConcurrentLinkedQueue：无阻塞无容量队列 ​

ConcurrentSkipListMap：并发有序 Map ​

LinkedTransferQueue.transfer：等待消费者接收 ​

keySet 视图：并发 Map 的集合视图 ​

简单案例 ​

易混点 ​

课后小问 ​

本节小结 ​

快速回顾 ​

并发集合与阻塞队列

学习目标

核心知识点

专业术语

白话解释与边界

常用用法

ConcurrentHashMap.putIfAbsent：只初始化一次

ConcurrentHashMap.computeIfAbsent：并发缓存

ConcurrentHashMap.compute：按旧值原子更新

ConcurrentHashMap.merge：累计或合并值

CopyOnWriteArrayList：读快照

阻塞式交接：放入并取出队列元素

有界队列交互：立即尝试放入或取出

LinkedBlockingQueue：可选容量的链式阻塞队列

不常用但需要知道

ConcurrentLinkedQueue：无阻塞无容量队列

ConcurrentSkipListMap：并发有序 Map

LinkedTransferQueue.transfer：等待消费者接收

`keySet` 视图：并发 Map 的集合视图

简单案例

易混点

课后小问

本节小结

快速回顾