并发编程 - 高性能队列 Disruptor

在各种各样的并发工具类中，如何确定我们代码的并发性能呢？这个时候就需要来聊下 Java 中提供的微基准测试工具 JMH, 它主要是基于方法层面的基准测试，精度可以达到纳秒级。JMH由Oracle内部实现JIT的大牛们编写，他们比任何人都了解JIT以及JVM对于基准测试的影响



# JMH （Java Microbenchmark Harness）

在现代Java开发中，性能优化不再是凭直觉猜测的游戏。JMH（[官网](http://openjdk.java.net/projects/code-tools/jmh/) ）作为由Oracle JIT团队开发的专业级微基准测试框架，提供了纳秒级精度的代码性能分析能力，成为衡量方法级性能的黄金标准。本文将全面解析JMH的核心原理、实践技巧与避坑指南。

## JMH 有什么用

JMH通过科学控制JVM变量（如JIT编译、垃圾回收）解决传统手工测试的三大痛点：

- 消除干扰因素：独立的测试进程隔离避免方法间优化干扰

- 预热机制：通过预热迭代触发JIT编译，使结果反映稳定性能

- 多维度统计：支持吞吐量(OPS)、平均时间、百分位数(TP99)等丰富指标

典型应用场景：

- 🔍 量化对比算法实现（如String拼接 vs StringBuilder）

- ⚡ 检测方法执行时间与输入规模的相关性

- 📊 验证性能优化效果（如缓存引入前后的吞吐量变化）

- 🧪 多线程并发场景下的性能压测

## JMH核心概念解剖

### 测试模式（Benchmark Mode）

| 模式	| 度量指标	| 适用场景| 
| --- | --- | --- |
| Throughput	| 每秒操作数(ops/s)	| 高并发接口吞吐评估| 
| AverageTime	| 单次操作平均耗时	| 算法效率对比| 
| SampleTime	| 响应时间分布(TP99)	| SLA合规性验证| 
| SingleShotTime	| 冷启动耗时	| 初始化性能测试| 

### 生命周期控制

**预热(Warmup)**：触发JIT编译，避免冷启动扭曲结果9

```java
@Warmup(iterations = 3, time = 1, timeUnit = TimeUnit.SECONDS)
```

**测量(Measurement)**：实际统计阶段

```java
@Measurement(iterations = 5, time = 5, timeUnit = TimeUnit.SECONDS)
```

### 状态作用域(State Scope)
| 作用域	| 实例分配规则	| 使用场景| 
| --- | --- | --- |
| Scope.Thread	| 每线程独立实例	| 无状态服务测试| 
| Scope.Benchmark	| 全局共享实例	| 单例组件性能| 
| Scope.Group	| 线程组共享实例	| 生产者-消费者模型| 

### 注解

因为我们主要利用JMH提供的注解来进行基准测试，因此我们有必要了解一下JMH一些常用注解

- @State: 表明类的所有属性的作用域。只能用于类上。它有如下选项

        Scope.Thread： 默认的State，每个测试线程分配一个实例；
        Scope.Benchmark： 所有测试线程共享一个实例，用于测试有状态实例在多线程共享下的性能；
        Scope.Group： 每个线程组共享一个实例；

- @BenchmarkMode: 用于指定基准测试的执行模式，如吞吐量、平均执行时间。可用于类或者方法上，它有如下模式

        Throughput：整体吞吐量，每秒执行了多少次调用，单位为 ops/time
        AverageTime：用的平均时间，每次操作的平均时间，单位为 time/op
        SampleTime：随机取样，最后输出取样结果的分布
        SingleShotTime：只运行一次，往往同时把 Warmup 次数设为 0，用于测试冷启动时的性能
        All：上面的所有模式都执行一次

- @Measurement: 用于控制压测的次数、时间和批处理数量。可用于类或者方法上，它有如下参数

        iterations：测量的次数
        time：每次测量持续的时间
        timeUnit：时间的单位，默认秒
        batchSize：批处理大小，每次操作调用几次方法

- @Warmup: 预热，可用于类或者方法上
        
        由于JVM会使用JIT对热点代码进行编译，因此同一份代码可能由于执行次数的增加而导致执行时间差异太大，因此我们可以让代码先预热几轮，预热时间不算入测量计时。@WarmUp 的使用和 @Measurement 一致。
        
- @Fork: 用于指定fork出多少个子进程来执行同一基准测试方法，可用于类或者方法上。例如@Fork指定数量为2，则 JMH 会 fork 出两个进程来进行测试

- @Threads: 用于指定使用多少个线程来执行基准测试方法，可用于类或者方法上。例如@Threads 指定线程数为 2 ，那么每次测量都会创建两个线程来执行基准测试方法

- @OutputTimeUnit: 可以指定输出的时间单位，可用于类或者方法注解

- @Param： 指定某项参数的多种情况，特别适合用来测试一个函数在不同的参数输入的情况下的性能，只能作用在字段上，使用该注解必须定义 @State 注解。

- @Setup： 用于基准测试前的初始化动作，只能用于方法

- @TearDown  用于基准测试后执行，主要用于资源的回收，只能用于方法


## SpringBoot 集成实践

### 依赖

添加需要添加两个依赖：jmh-core （jmh的核心)、jmh-generator-annprocess（注解处理包） 

最新版本可以在Maven仓库 https://mvnrepository.com/ 中找到。

```xml
<properties>
    <jmh.version>1.36</jmh.version>
</properties>

<dependencies>
    <dependency>
        <groupId>org.openjdk.jmh</groupId>
        <artifactId>jmh-core</artifactId>
        <version>${jmh.version}</version>
    </dependency>
    <dependency>
        <groupId>org.openjdk.jmh</groupId>
        <artifactId>jmh-generator-annprocess</artifactId>
        <version>${jmh.version}</version>
        <scope>provided</scope>
    </dependency>
</dependencies>
```

```gradle
// Gradle 示例
dependencies {
    implementation 'org.openjdk.jmh:jmh-core:LATEST_VERSION'
    annotationProcessor 'org.openjdk.jmh:jmh-generator-annprocess:LATEST_VERSION'
}
```

### JMH 生态

JMH 是 OpenJDK 项目的一部分，广泛用于 Java 性能测试。以下是一些与 JMH 相关的项目和工具：

- JMH Visualizer：一个用于可视化 JMH 测试结果的工具。
- jmh-gradle-plugin：一个用于在 Gradle 项目中集成 JMH 的插件。
- jmh-compare-gui：一个用于比较不同 JMH 测试结果的图形界面工具。

通过这些工具和插件，可以更全面地进行 Java 性能测试和分析。

Idea 中的 [JMH Java Microbenchmark Harness 插件官网](https://plugins.jetbrains.com/plugin/7529-jmh-java-microbenchmark-harness)

JMH 在去使用的时候，应该将项目构建成jar包，然后在服务器上进行微基准测试，服务器性能上约接近生产越好，如果测试性能完全ok，那到生产服务器上也大致相同。

在开发的过程中，如果说边开发边进行这种微基准的测试实际上是不准确的，毕竟开发环境会对结果产生影响。平时你要想进行一些微基准的测试的话，要是每次打个包来进行正规一个从头到尾的测试 ，完了之后发现问题不对再去重新改，效率太低了，这个时候就需要用到我们的插件。


## JMH 测试陷阱

### 死代码消除

运行微基准测试时，了解优化非常重要。否则，它们可能会以非常误导的方式影响基准测试结果。

为了使事情更具体一些，让我们考虑一个例子：

```java
@Benchmark
@OutputTimeUnit(TimeUnit.NANOSECONDS)
@BenchmarkMode(Mode.AverageTime)
public void doNothing() {
}

@Benchmark
@OutputTimeUnit(TimeUnit.NANOSECONDS)
@BenchmarkMode(Mode.AverageTime)
public void objectCreation() {
    new Object();
}
```

我们期望对象分配的成本高于什么都不做。但是，如果我们运行基准测试：

```text
Benchmark                 Mode  Cnt  Score   Error  Units
BenchMark.doNothing       avgt   40  0.609 ± 0.006  ns/op
BenchMark.objectCreation  avgt   40  0.613 ± 0.007  ns/op
```

显然，在 TLAB 中找到一个位置，创建和初始化一个对象几乎是免费的！仅通过查看这些数字，我们应该知道这里有些东西并没有完全加起来。

在这里，我们是死代码消除的受害者。编译器非常擅长优化冗余代码。事实上，这正是 JIT 编译器在这里所做的。

为了防止这种优化，我们应该以某种方式欺骗编译器并使其认为代码被其他组件使用。 实现此目的的一种方法是返回创建的对象：

```java
@Benchmark
@OutputTimeUnit(TimeUnit.NANOSECONDS)
@BenchmarkMode(Mode.AverageTime)
public Object pillarsOfCreation() {
    return new Object();
}
```

此外，我们可以让黑洞消耗它：

```java
@Benchmark
@OutputTimeUnit(TimeUnit.NANOSECONDS)
@BenchmarkMode(Mode.AverageTime)
public void blackHole(Blackhole blackhole) {
    blackhole.consume(new Object());
}
```

让黑洞使用对象是说服 JIT 编译器不应用死代码消除优化的一种方法。无论如何，如果我们再次运行这些基准测试，这些数字将更有意义：

```text
Benchmark                    Mode  Cnt  Score   Error  Units
BenchMark.blackHole          avgt   20  4.126 ± 0.173  ns/op
BenchMark.doNothing          avgt   20  0.639 ± 0.012  ns/op
BenchMark.objectCreation     avgt   20  0.635 ± 0.011  ns/op
BenchMark.pillarsOfCreation  avgt   20  4.061 ± 0.037  ns/op
```

### 恒定折叠

让我们考虑另一个例子：

```java
@Benchmark
public double foldedLog() {
    int x = 8;

    return Math.log(x);
}
```

基于常量的计算可能会返回完全相同的输出，而不管执行次数如何。 因此，JIT 编译器很有可能将对其结果替换对数函数调用：

```java
@Benchmark
public double foldedLog() {
    return 2.0794415416798357;
}
```

这种形式的部分评估称为恒定折叠。在这种情况下，不断折叠完全避免了 Math.log 调用，这是基准测试的重点。

为了防止常量折叠，我们可以将常量状态封装在一个状态对象中：

```java
@State(Scope.Benchmark)
public static class Log {
    public int x = 8;
}

@Benchmark
public double log(Log input) {
     return Math.log(input.x);
}
```

如果我们相互运行这些基准：

```text
Benchmark             Mode  Cnt          Score          Error  Units
BenchMark.foldedLog  thrpt   20  449313097.433 ± 11850214.900  ops/s
BenchMark.log        thrpt   20   35317997.064 ±   604370.461  ops/s
```

## 创建JMH测试

### 字符串拼接性能对决

测试+拼接 vs StringBuilder在不同数据量下的性能：

```java
@State(Scope.Benchmark)
@BenchmarkMode(Mode.AverageTime)
@OutputTimeUnit(TimeUnit.NANOSECONDS)
public class StringConnectTest {
    @Param({"10", "100", "1000"})
    private int length;

    @Benchmark
    public void testStringAdd(Blackhole bh) {
        String a = "";
        for (int i = 0; i < length; i++) {
            a += i;
        }
        bh.consume(a);
    }

    @Benchmark
    public void testStringBuilder(Blackhole bh) {
        StringBuilder sb = new StringBuilder();
        for (int i = 0; i < length; i++) {
            sb.append(i);
        }
        bh.consume(sb.toString());
    }
}
```

典型结果分析

```text
Benchmark               (length)   Mode   Cnt     Score    Error  Units
testStringAdd               10     avgt    5     161.5 ±  17.1  ns/op
testStringBuilder           10     avgt    5      34.2 ±   0.7  ns/op
testStringAdd              100     avgt    5   14768.9 ± 452.3  ns/op
testStringBuilder          100     avgt    5     317.5 ±   5.2  ns/op
```

数据表明：超过100次拼接时，StringBuilder比+快46倍以上，也就是指标 `14768.9 / 317.5 = 46`, 揭示性能差距随数据量指数级扩大。


### 测试Spring Bean性能

通过@Setup初始化Spring上下文，实现真实环境下的组件性能验证

```java
@State(Scope.Benchmark)
public class ServiceBenchmark {
    private ConfigurableApplicationContext context;
    private UserService userService;

    @Setup
    public void init() {
        context = SpringApplication.run(MainApp.class);
        userService = context.getBean(UserService.class);
    }

    @Benchmark
    public void testUserQuery() {
        userService.findById(1L); // 测试DAO方法性能
    }

    @TearDown
    public void close() {
        context.close();
    }
}
```

:::tip{title="运行测试类，如果遇到下面的错误"}
```java
ERROR: org.openjdk.jmh.runner.RunnerException: ERROR: Exception while trying to acquire the JMH lock (C:\WINDOWS\/jmh.lock): C:\WINDOWS\jmh.lock (拒绝访问。), exiting. Use -Djmh.ignoreLock=true to forcefully continue.
    at org.openjdk.jmh.runner.Runner.run(Runner.java:216)
    at org.openjdk.jmh.Main.main(Main.java:71)
```

这个错误是因为JMH运行需要访问系统的TMP目录，解决办法是：打开 `RunConfiguration -> Environment Variables -> include system environment viables`
:::


### 官方样例

如果说大家对JMH有兴趣，你们在工作中可能会有用的上大家去读一下 [官方](http://hg.openjdk.java.net/code-tools/jmh/file/tip/jmh-samples/src/main/java/org/openjdk/jmh/samples/) 的例子，官方大概有好几十个例子程序，你可以自己一个一个的去研究。

## 总结

JMH 将性能测试从“经验猜测”提升到科学度量层面，使用时需注意：

测试策略：

- 优先使用Throughput模式评估服务容量

- 用SampleTime分析长尾请求

配置原则：

- 预热迭代≥3次，持续≥1秒19

- 测量迭代≥5次确保统计显著性

结果解读：

- 关注Score而非绝对时间

- 误差值(Error)超过10%需增加测试迭代7

性能优化应遵循 **“测量-优化-验证”** 闭环。JMH提供了这个闭环中最可靠的测量工具，让每一次优化都有据可循。真正的性能洞察始于精确测量，而非直觉猜测。掌握JMH，让性能调优从玄学走向科学。

性能分析 - JMH 微基准测试

并发编程 - 深入解析 ConcurrentHashMap

你可能对List/Set/Map有了解，那juc包下的容器与队列呢？本文将全面剖析 Java 并发容器体系，涵盖从传统同步容器到现代高性能并发容器的演进，结合代码示例与性能对比，帮助你构建高并发体系知识




## 同步容器（JDK 1.0 - JDK 1.2）

### 核心类与接口

| 类/接口 | 描述 | 线程安全实现方式 |
| --- | --- | --- |
| Vector	| 同步版 ArrayList	| 所有方法用 synchronized| 
| Hashtable	| 同步版 HashMap	| 所有方法用 synchronized| 
| Stack	 | 同步栈（继承 Vector）	| 继承 Vector 的同步 | 
| Collections.synchronizedList(List)	| 包装普通 List 为同步 List	| 内部锁对象同步| 
| Collections.synchronizedMap(Map)	| 包装普通 Map 为同步 Map	| 内部锁对象同步| 
| Collections.synchronizedSet(Set)	| 包装普通 Set 为同步 Set	| 内部锁对象同步| 


### 代码示例：同步容器基本使用

```java
// Vector 使用示例
Vector<String> vector = new Vector<>();
vector.add("item1");
vector.remove(0);

// synchronizedList 使用
List<String> syncList = Collections.synchronizedList(new ArrayList<>());
syncList.add("safe");

// 迭代时需要手动同步！
synchronized(syncList) {
    Iterator<String> it = syncList.iterator();
    while(it.hasNext()) {
        System.out.println(it.next());
    }
}
```

### 同步容器缺陷分析

性能瓶颈：全表锁导致高并发下性能差

```java
// Hashtable 的 put 方法（全表锁）
public synchronized V put(K key, V value) { ... }
```

复合操作非原子性：
```java
// 错误示例：先检查再操作
if(!vector.contains(item)) {
    vector.add(item); // 可能被其他线程插入
}
```

迭代器弱一致性：
```java
// ConcurrentModificationException 风险
for(String s : vector) {
    vector.remove(s); // 运行时异常
}
```


:::warning{title="注意"}
现阶段，几乎没有对同步容器的任何应用了
:::

## 并发容器（JUC 框架）

### 并发 Map 实现


| 类名 | 数据结构 | 特点 | JDK 版本 |
| --- | --- | --- | --- |
| ConcurrentHashMap	| 数组+链表+红黑树	| 分段锁/CAS	| 1.5+| 
| ConcurrentSkipListMap	| 跳表	| 天然有序/无锁读	| 1.6+| 

```java
// ConcurrentHashMap 基本操作
ConcurrentHashMap<String, Integer> map = new ConcurrentHashMap<>();
map.put("a", 1);
map.compute("a", (k, v) -> v + 1); // 原子更新

// 高并发统计（线程安全）
map.merge("count", 1, Integer::sum);

// 统计单词频率（线程安全）
ConcurrentHashMap<String, LongAdder> freqMap = new ConcurrentHashMap<>();
words.forEach(word -> 
    freqMap.computeIfAbsent(word, k -> new LongAdder()).increment()
);
```


#### Map实现性能对比（百万次操作）

测试环境：8核 CPU，JDK 17

| 实现类	| 读操作(ms)	| 写操作(ms)	| 读写混合(ms)| 
| --- | --- | --- | --- | 
| Hashtable	| 120	| 450	| 320| 
| Collections.synchronizedMap	| 115	| 430	| 310| 
| ConcurrentHashMap	| 45	| 85	| 65| 
| ConcurrentSkipListMap	| 60	| 210	| 130| 


### 并发 List/Set 实现

|类名| 实现原理|适用场景| 
| --- | --- | --- |
| CopyOnWriteArrayList	| 写时复制数组	| 读多写少（监听器列表）| 
| CopyOnWriteArraySet	| 基于 CopyOnWriteArrayList	| 读多写少的小规模集合| 
| ConcurrentSkipListSet	| 基于 ConcurrentSkipListMap	| 有序且并发访问的集合| 

```java
// CopyOnWriteArrayList 示例
CopyOnWriteArrayList<String> logList = new CopyOnWriteArrayList<>();

// 写操作（复制整个数组）
logList.add("Log entry: " + System.currentTimeMillis());

// 读操作（无锁访问原始数组）
for(String log : logList) { // 迭代器访问快照
    System.out.println(log);
}
```

## 阻塞队列（BlockingQueue）


### 阻塞队列的方法
 
**生产方法**
  
- add(E) 添加数据到队列，如果队列满了，抛出异常，添加成功返回true
- offer(E) 添加数据到队列，如果队列满了，返回false，添加成功返回true
- offer(E,timeout,unit) 添加数据到队列，如果队列满了，阻塞timeout时间，到期后还不能添加成功，返回false，添加成功返回true
- put(E) 添加数据到队列，如果队列满了，挂起线程，一直等到队列中有位置，再扔数据进去，死等！
 
**消费方法**
  
- remove() 从队列中移除数据，如果队列为空，抛出异常
- poll() 从队列中移除数据，如果队列为空，返回null，么的数据
- poll(timeout,unit) 从队列中移除数据，如果队列为空，挂起线程timeout时间，等生产者扔数据，再获取
- take() 从队列中移除数据，如果队列为空，线程挂起，一直等到生产者扔数据，再获取
 
**检查方法**

- element() 检索但不删除此队列的头部，如果队列如空，则抛出异常
- peek() 检索但不删除此队列的头部，如果队列如空，则返回null


### 阻塞队列的实现

| 队列实现类	| 数据结构	| 特点| 
| --- | --- | --- |
| ArrayBlockingQueue	| 数组	| 有界/FIFO/固定容量| 
| LinkedBlockingQueue	| 链表	| 可选有界/FIFO/高吞吐| 
| PriorityBlockingQueue	| 堆	| 无界/优先级排序| 
| DelayQueue	| 优先级堆	| 元素按延迟时间出队| 
| SynchronousQueue	| 无缓冲	| 直接传递（生产者-消费者）| 
| LinkedTransferQueue	| 链表	| 融合阻塞与非阻塞特性| 

#### ArrayBlockingQueue

- 底层基于数组，是一个有界的先进先出（FIFO）的队列
- 队列的头部是队列中时间最长的元素。队列的尾部是队列中时间最短的元素。新元素插入队列的尾部，队列检索操作获取队列头部的元素
- 方法内部使用 ReentrantLock 来保证线程安全
- 索引
      
      putIndex:  下一个 put、offer 或 add 的项目索引
      takeIndex:  下次take, poll, peek 或 remove 的项目索引

**代码示例**

```java
// 生产者
BlockingQueue<Task> queue = new ArrayBlockingQueue<>(100);
queue.put(task); // 阻塞直到空间可用

// 消费者
Task task = queue.take(); // 阻塞直到元素可用

// 带超时操作
boolean success = queue.offer(task, 1, TimeUnit.SECONDS);
```

#### LinkedBlockingQueue
 
- 底层是基于单向链表，是一个默认无界的先进先出（FIFO）的队列
-  链表队列通常比基于数组的队列具有更高的吞吐量
- 默认容量长度是 Integer.MAX
  ```java
    public LinkedBlockingQueue() {
        this(Integer.MAX_VALUE);
    }
    ```
- 链表节点
  ```java
  static class Node<E> {
    E item;
    Node<E> next;
    Node(E x) { item = x; }
  }
  ```
- 使用2把 ReentrantLock 分别来控制入队与出队的线程安全

#### PriorityBlockingQueue

- 基于二叉堆（二叉堆底层基于数组实现）是一个优先级阻塞队列，默认数组长度是11
- 没有自定义比较器，默认基于自然排序
- 方法内部使用 ReentrantLock 来保证线程安全
- 入队的对象必须实现 `Comparator<? super E>` 接口，或者创建 PriorityBlockingQueue 对象时必须指定  `Comparator<? super E>`

#### DelayQueue
 
- 无界的延迟阻塞队列，其中元素只有在其延迟过期时才能出队
- 入队的对象必须实现 Delayed 接口 - `DelayQueue<E extends Delayed>`
    ```java
    interface Delayed extends Comparable<Delayed>
    ```
 - 内部基于 PriorityQueue 对时间进行排序存储
 - 无界的优先级队列（非阻塞队列 - 没有put、poll( timeunit )、take方法）
 - 底层基于二叉堆实现 - 实际是数组
    ```java
    public class PriorityQueue<E> extends AbstractQueue<E>
    ``` 
 - 在比较时，跟根据延迟时间做比较，剩余时间最短的要放在栈顶
-  方法内部使用 ReentrantLock 来保证线程安全

**延迟队列实现定时任务：**

```java
DelayQueue<DelayedTask> delayQueue = new DelayQueue<>();
delayQueue.put(new DelayedTask("task1", 5000)); // 5秒后执行

while(true) {
    DelayedTask task = delayQueue.take(); // 阻塞直到到期
    task.execute();
}
```

#### SynchronousQueue

- SynchronousQueue不存储任何元素
- 直接传递（生产者-消费者）
 
#### LinkedTransferQueue

- 实现了TransferQueue接口，而TransferQueue接口是继承自BlockingQueue的
- 使用了一个叫做 dual data structure的数据结构，或者叫做 dual queue，译为双重数据结构或者双重队列。

**TransferQueue 实现消息确认：**

```java
TransferQueue<Message> xferQueue = new LinkedTransferQueue<>();

// 生产者（等待消费者接收）
xferQueue.transfer(message); 

// 消费者
Message msg = xferQueue.take();
process(msg);
```

## 非阻塞队列

| 队列实现类	| 数据结构	| 特点| 
| --- | --- | --- |
| ConcurrentLinkedQueue	| 链表	| CAS 无锁/FIFO/高并发| 
| ConcurrentLinkedDeque	| 双向链表	| 双端无锁操作| 

```java
// 高性能并发队列
ConcurrentLinkedQueue<String> clq = new ConcurrentLinkedQueue<>();
clq.offer("event");
String event = clq.poll();
```

## 双端队列（Deque）

| 实现类	| 阻塞类型	| 特点| 
| --- | --- | --- |
| ArrayDeque	| 非阻塞	| 数组实现/高效随机访问| 
| LinkedBlockingDeque	| 阻塞	| 链表实现/容量可选| 
| ConcurrentLinkedDeque	| 非阻塞	| 单向链表无界 CAS 无锁实现| 

```java
// 双端队列操作
Deque<String> deque = new LinkedBlockingDeque<>(10);
deque.offerFirst("urgent"); // 队首插入
deque.offerLast("normal");  // 队尾插入
String item = deque.pollFirst(); // 队首获取
```

## 队列吞吐量对比（每秒操作数）

测试环境：8核 CPU，JDK 17

| 队列类型	| 1线程	| 4线程	| 8线程| 
| --- |  --- | --- | --- | 
| ArrayBlockingQueue	| 150,000	| 280,000	| 320,000| 
| LinkedBlockingQueue	| 180,000	| 550,000	| 900,000| 
| ConcurrentLinkedQueue	| 250,000	| 980,000	| 1,200,000| 
| SynchronousQueue	| 140,000	| 480,000	| 750,000| 

## 最佳实践与选型指南

### 容器选型决策树

![deepseek_mermaid_20250606_97aad2.png](https://blog.liushigong.cn/static/img/bf97541db33de90d65fbee42e1b973b4.deepseek_mermaid_20250606_97aad2.webp)

### 使用禁忌与陷阱

1. 错误： 在迭代中修改 CopyOnWriteArrayList

    ```java
    // 正确：迭代器访问的是快照
    for(String item : cowList) {
        cowList.remove(item); // 允许但不影响当前迭代
    }
    ```

2. 注意： ConcurrentHashMap 的弱一致性迭代

    ```java
    ConcurrentHashMap<String, Integer> map = ...;
    // 不保证迭代时看到最新修改
    map.forEach((k, v) -> System.out.println(k));
    ```

3. 避免： 无界队列导致 OOM

    ```java
    // 危险：默认无界可能撑爆内存
    BlockingQueue<BigData> queue = new PriorityBlockingQueue<>();
    // 推荐：使用有界队列
    BlockingQueue<BigData> safeQueue = new ArrayBlockingQueue<>(1000);
    ```
    
## 总结

| 队列实现类	| 数据结构	| 并发实现 |  特点 |
| --- | --- | --- | --- | 
| ConcurrentHashMap	| 数组+链表+红黑树	| 分段锁/CAS	|  并发map，性能极好 |
| ConcurrentSkipListMap	| 跳表	            | 天然有序/无锁读	 |  有序map |
| CopyOnWriteArrayList	| 写时复制数组	 | synchronized + volatile 数组      | 有序集合 |
| CopyOnWriteArraySet	| 数组	| 基于 CopyOnWriteArrayList | 基于数组的无序集合 |
| ConcurrentSkipListSet	| 链表 | 基于 ConcurrentSkipListMap 	| 有序且并发访问的集合| | 
| ArrayBlockingQueue	| 数组	| ReentrantLock + Condition | 有界/FIFO/固定容量| 
| LinkedBlockingQueue	| 链表	| ReentrantLock + Condition | 可选有界/FIFO/高吞吐| 
| PriorityBlockingQueue	| 堆	| ReentrantLock + Condition | 无界/优先级排序| 
| PriorityQueue | 二叉堆（数组） | 不是线程安全的 | 有序 | 
| DelayQueue	| 基于PriorityQueue	| ReentrantLock | 元素按延迟时间出队| 
| SynchronousQueue	| 无缓冲  | CAS	| 直接传递（生产者-消费者）| 
| LinkedTransferQueue	| 链表	| CAS | 融合阻塞与非阻塞特性| 
| ConcurrentLinkedQueue	| 链表	| CAS | CAS 无锁/FIFO/高并发| 
| ArrayDeque	| 数组实现	| 不是线程安全的 | 高效随机访问| 
| LinkedBlockingDeque	| 链表实现	| ReentrantLock | 容量可选| 
| ConcurrentLinkedDeque	| 双向链表|  CAS 无锁实现| 双端无锁操作 |

- 读多写少 → CopyOnWriteArrayList
- KV 存储 → ConcurrentHashMap
- 有序映射 → ConcurrentSkipListMap
- 任务调度 → PriorityBlockingQueue
- 高吞吐传输 → LinkedTransferQueue
- 极低延迟 → ConcurrentLinkedDeque

通过合理选择并发容器，可提升系统并发能力 3-10 倍，同时降低死锁风险。理解各容器实现原理是构建高性能 Java 应用的基石。

并发编程 - 并发容器全解析

我们都知道在高并发下 `ArrayList` 是线程不安全的，那如果要使用集合要如何处理呢？接下来即将深入探讨Java并发容器中的 `CopyOnWriteArrayList`。见名知义，包含 `Array` 的底层还是通过数组实现的，这种特别适合读多写少的并发环境，接下来我们去深入了解下 `CopyOnWriteArrayList` 是如何能在高并发下保证数据的一致性的。




## CopyOnWriteArrayList

`CopyOnWriteArrayList`（COW）是一个线程安全的 `ArrayList`，采用 **"写时复制"** 思想，实现了读操作完全无锁的高性能并发容器。


### 核心设计思想

`CopyOnWriteArrayList` 是基于 `ReentrantLock` 锁和 **创建数组副本** 的形式去保证线程安全。

**读取**
读取：直接访问当前数组，无需加锁


**写入**

在写数据时，先获取 `lock` 锁，复制原数组创建一个副本数组，在新数组上执行修改操作，将副本数组赋值给`CopyOnWriteArrayList` 中的 `array`（因为是 `volatile` 的，原子替换引用指向新数组）。


**写操作性能开销（内存复制 + GC压力）**

因为 `CopyOnWriteArrayList` 每次写数据都要构建一个副本，如果你的业务是写多，并且数组中的数据量比较大，尽量避免去使用 `CopyOnWriteArrayList`，因为这里会构建大量的数组副本，比较占用内存资源。

`CopyOnWriteArrayList` 是弱一致性的，写操作先执行，在副本还没有落到 `CopyOnWriteArrayList` 的`array` 属性中，此时读操作是无法查询到的。

```java
// 伪代码演示COW机制
public void add(E element) {
    Object[] oldArray = currentArray;        // 1. 获取当前数组
    Object[] newArray = Arrays.copyOf(oldArray, oldArray.length + 1); // 2. 复制新数组
    newArray[newArray.length - 1] = element; // 3. 修改新数组
    currentArray = newArray;                 // 4. 原子替换引用
}
```

### 源码深度解析

主要查看2个核心属性，以及2个核心方法，还有无参构造

```java
/** 写操作时，需要先获取到的锁资源，CopyOnWriteArrayList 全局唯一的。 */
final transient ReentrantLock lock = new ReentrantLock();

/** 保证可见性的数组引用 （volatile保证写操作对其他线程立即可见）CopyOnWriteArrayList 真实存放数据的位置，查询也是查询当前array */
private transient volatile Object[] array;

// 获取当前数组属性
final Object[] getArray() {
    return array;
}

// 替换array属性
final void setArray(Object[] a) {
    array = a;
}

/**
 *  默认new的 CopyOnWriteArrayList 数组长度为0。
 *  不像ArrayList，初始长度是10，每次扩容1/2, CopyOnWriteArrayList不存在这个概念
 *  每次写的时候都会构建一个新的数组
 */
public CopyOnWriteArrayList() {
    setArray(new Object[0]);
}
```

:::tip{title="volatile + 锁的双重保障"}
- volatile 保证数组引用的可见性
- ReentrantLock 保证写操作的原子性
:::

#### 读操作

`CopyOnWriteArrayList` 的读操作就是 `get` 方法，基于数组索引位置获取数据。

方法之所以要差分成两个，是因为 `CopyOnWriteArrayList` 中在获取数据时，不单单只有一个 `array` 的数组需要获取值，还有副本中数据的值。

```java
// 查询数据时，只能通过get方法查询CopyOnWriteArrayList中的数据
public E get(int index) {
    // 直接访问当前数组，调用get方法的重载
    return get(getArray(), index);
}
// 执行get(int)时，内部调用的方法
private E get(Object[] a, int index) {
    // 直接拿到数组上指定索引位置的值，无锁读取数据
    return (E) a[index];
}
```

#### 写操作

`CopyOnWriteArrayList` 是基于 `lock` 锁和 **数组副本** 的形式保证线程安全。

```java
// 写入元素，不指定索引位置，直接放到最后的位置
public boolean add(E e) {
    // 获取全局锁，并执行lock
    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;
        // 将副本数组，赋值给CopyOnWriteArrayList的原数组
        setArray(newElements);
        // 添加成功，返回true
        return true;
    } finally {
        // 释放锁~
        lock.unlock();
    }
}

// 写入元素，指定索引位置。（不会覆盖数据）
public void add(int index, E element) {
    // 拿锁，加锁~
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        // 获取原数组，还获取了原数组的长度
        Object[] elements = getArray();
        int len = elements.length;
        // 如果索引位置大于原数组的长度，或者索引位置是小于0的。
        if (index > len || index < 0)
            throw new IndexOutOfBoundsException("Index: "+index+
                                                ", Size: "+len);
        // 声明了副本数组
        Object[] newElements;
        // 原数组长度 - 索引位置等到numMoved
        int numMoved = len - index;
        // 如果numMoved为0，说明数据要放到最后面的位置
        if (numMoved == 0)
            // 直接走了原生态的方式，正常复制一份副本数组
            newElements = Arrays.copyOf(elements, len + 1);
        else {
            // 数组要插入的位置不是最后一个位置
            // 副本数组长度依然是原长度 + 1
            newElements = new Object[len + 1];
            // 将原数组从0索引位置开始复制，复制到副本数组中的前置位置
            System.arraycopy(elements, 0, newElements, 0, index);
            // 将原数组从index位置开始复制，复制到副本数组的index + 1往后放。
            // 这时，index就空缺出来了。
            System.arraycopy(elements, index, newElements, index + 1,
                             numMoved);
        }
        // 数据正常放到指定的索引位置
        newElements[index] = element;
        // 将副本数组，赋值给CopyOnWriteArrayList的原数组
        setArray(newElements);
    } finally {
        // 释放锁
        lock.unlock();
    }
}
```

#### 移除数据

关于 `remove` 操作，要分析两个方法

* 基于索引位置移除指定数据
* 基于具体元素删除数组中最靠前的数据
    1. 当前这种方式，嵌套了一层，导致如果元素存在话，成本是比较高的。
    2. 如果元素不存在，这种设计不需要加锁，提升写的效率

```java
// 删除指定索引位置的数据
public E remove(int index) {
    // 拿锁，加锁
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        // 获取原数组和原数组长度
        Object[] elements = getArray();
        int len = elements.length;
        // 通过get方法拿到index位置的数据
        E oldValue = get(elements, index);
        // 声明numMoved
        int numMoved = len - index - 1;
        // 如果numMoved为0，说明删除的元素是最后的位置
        if (numMoved == 0)
            // Arrays.copyOf复制一份新的副本数组，并且将最后一个数据不要了
            // 基于setArray将副本数组赋值给array原数组
            setArray(Arrays.copyOf(elements, len - 1));
        else {
            // 删除的元素不在最后面的位置
            // 声明副本数组，长度是原数组长度 - 1
            Object[] newElements = new Object[len - 1];
            // 从0开始复制的index前面
            System.arraycopy(elements, 0, newElements, 0, index);
            // 从index后面复制到最后
            System.arraycopy(elements, index + 1, newElements, index,
                             numMoved);
            setArray(newElements);
        }
        // 返回被干掉的数据
        return oldValue;
    } finally {
        // 释放锁
        lock.unlock();
    }
}

// 删除元素（最前面的）
public boolean remove(Object o) {
    // 没加锁！！！！
    // 获取原数组
    Object[] snapshot = getArray();
    // 用indexOf获取元素在数组的哪个索引位置
    // 没找到的话，返回-1
    int index = indexOf(o, snapshot, 0, snapshot.length);
    // 如果index < 0,说明元素没找到，直接返回false，告辞
    // 如果找到了元素的位置，直接执行remove方法的重载
    return (index < 0) ? false : remove(o, snapshot, index);
}

// 执行remove(Object o)，找到元素位置时，执行当前方法
private boolean remove(Object o, Object[] snapshot, int index) {
    // 拿锁，加锁
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        // 拿到原数组和长度
        Object[] current = getArray();
        int len = current.length;
        // findIndex: 是给if起标识，break 标识的时候，直接跳出if的代码块~~
        if (snapshot != current) findIndex: {
            // 如果没进到if，说明数组没变化，按照原来的index位置删除即可
            // 进到这，说明数组有变化，之前的索引位置不一定对
            // 拿到index位置和原数组长度的值
            int prefix = Math.min(index, len);
            // 循环判断，数组变更后，是否影响到了要删除元素的位置
            for (int i = 0; i < prefix; i++) {
                // 如果不相等，说明元素变化了。
                // 同时判断变化的元素是否是我要删除的元素o
                if (current[i] != snapshot[i] && eq(o, current[i])) {
                    // 如果满足条件，说明当前位置就是我要删除的元素
                    index = i;
                    break findIndex;
                }
            }
            // 如果for循环结束，没有退出if，说明元素可能变化了，总之没找到要删除的元素
            // 如果删除元素的位置，已经大于等于数组长度了。
            if (index >= len)
                // 超过索引范围了，没法删除了。
                return false;
            // 索引还在范围内，判断是否是还是原位置，如果是，直接跳出if代码块
            if (current[index] == o)
                break findIndex;
            // 重新找元素在数组中的位置
            index = indexOf(o, current, index, len);
            // 找到直接跳出if代码块
            // 没找到。执行下面的return false
            if (index < 0)
                return false;
        }
        // 删除套路，构建新数组，复制index前的，复制index后的
        Object[] newElements = new Object[len - 1];
        System.arraycopy(current, 0, newElements, 0, index);
        System.arraycopy(current, index + 1,
                         newElements, index,
                         len - index - 1);
        // 复制到array
        setArray(newElements);
        // 返回true，删除成功
        return true;
    } finally {
        lock.unlock();
    }
}
```

#### 覆盖数据 & 清空集合

覆盖数据就是 `set` 方法，可以将指定位置的数据替换

```java
// 覆盖数据
public E set(int index, E element) {
    // 拿锁，加锁
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        // 获取原数组
        Object[] elements = getArray();
        // 获取原数组的原位置数据
        E oldValue = get(elements, index);

        // 原数据和新数据不一样
        if (oldValue != element) {
            // 拿到原数据的长度，复制一份副本。
            int len = elements.length;
            Object[] newElements = Arrays.copyOf(elements, len);
            // 将element替换掉副本数组中的数据
            newElements[index] = element;
            // 写回原数组
            setArray(newElements);
        } else {
            // 原数据和新数据一样，啥不干，把拿出来的数组再写回去
            setArray(elements);
        }
        // 返回原值
        return oldValue;
    } finally {
        // 释放锁
        lock.unlock();
    }
}
```

清空就是清空了~~~

```java
public void clear() {
	// 加锁
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
		// 扔一个长度为0的数组
        setArray(new Object[0]);
    } finally {
        lock.unlock();
    }
}
```

#### 迭代器

用 `ArrayList` 时，如果想在遍历的过程中去移除或者修改元素，必须使用迭代器才可以。

即使写操作中原始数组被替换，迭代器仍访问旧数组，不让在迭代时做写操作是因为不希望迭代操作时，会影响到写操作，还有，不希望迭代时，还需要加锁。

```java
// 获取遍历CopyOnWriteArrayList的iterator。
public Iterator<E> iterator() {
    // 其实就是new了一个COWIterator对象，并且获取了array，指定从0开始遍历
    return new COWIterator<E>(getArray(), 0);
}

static final class COWIterator<E> implements ListIterator<E> {
    /** 遍历的快照 */
    private final Object[] snapshot;
    /** 游标，索引~~~ */
    private int cursor;

    // 有参构造
    private COWIterator(Object[] elements, int initialCursor) {
        cursor = initialCursor;
        snapshot = elements;
    }

    // 有没有下一个元素，基于遍历的索引位置和数组长度查看
    public boolean hasNext() {
        return cursor < snapshot.length;
    }
    // 有没有上一个元素
    public boolean hasPrevious() {
        return cursor > 0;
    }

    // 获取下一个值，游标动一下
    public E next() {
        // 确保下个位置有数据
        if (! hasNext())
            throw new NoSuchElementException();
        return (E) snapshot[cursor++];
    }

    // 获取上一个值，游标往上移动
    public E previous() {
        if (! hasPrevious())
            throw new NoSuchElementException();
        return (E) snapshot[--cursor];
    }

    // 拿到下一个值的索引，返回游标
    public int nextIndex() {
        return cursor;
    }

    // 拿到上一个值的索引，返回游标
    public int previousIndex() {
        return cursor-1;
    }

    // 写操作全面禁止！！
    public void remove() {
        throw new UnsupportedOperationException();
    }

  
    public void set(E e) {
        throw new UnsupportedOperationException();
    }

  
    public void add(E e) {
        throw new UnsupportedOperationException();
    }

    // 兼容函数式编程
    @Override
    public void forEachRemaining(Consumer<? super E> action) {
        Objects.requireNonNull(action);
        Object[] elements = snapshot;
        final int size = elements.length;
        for (int i = cursor; i < size; i++) {
            @SuppressWarnings("unchecked") E e = (E) elements[i];
            action.accept(e);
        }
        cursor = size;
    }
}
```


## 总结

`CopyOnWriteArrayList` 是 Java 并发容器的独特存在，其优势在读操作完全无锁，适用于 **读:写 > 100:1** 的场景，由于是数组实现，在写操作上内存开销大，不适合频繁修改或大数据集，需要在享受 **COW** 的无锁读取红利时，务必警惕其写入代价。理解其底层复制机制，方能发挥最大威力！

并发编程 - CopyOnWriteArrayList 源码分析

baidu-site-verification

本人自己的博客，对工作与学习过程中的总结、一些场景设计的实现方案、遇到过的疑难问题与解决、以及面试总结等。