Java多线程进阶与JUC中线程安全的集合

约 4043 个字 79 行代码 预计阅读时间 14 分钟

常见锁策略

悲观锁与乐观锁

在Linux线程安全与死锁中已经对悲观锁和乐观锁有了概念介绍，此处不再说明

在Java中，synchronized使用的锁是悲观锁和乐观锁的结合，根据实际锁的竞争情况来进行切换，当竞争强烈时就会使用悲观锁

重量级锁和轻量级锁

• 重量级锁：加锁机制高度依赖操作系统提供的锁机制
• 轻量级锁：加锁机制尽可能不使用操作系统提供的锁机制，而是尽量在用户态完成。当无法处理时，会切换为重量级锁

在Java中，synchronized最开始为轻量级锁，如果锁竞争严重，则会切换为重量级锁

挂起等待锁和自旋锁

• 挂起等待锁：线程抢锁失败后，不继续自旋空转，而是被JVM/操作系统挂起，进入阻塞或挂起状态，等锁可用后再被唤醒继续竞争
• 自旋锁：线程抢锁失败后，它们会持续自旋（即在一个循环中不断检查锁是否可用）而不是立即进入休眠状态等待锁的释放

在Java中，synchronized的轻量级锁策略大概率就是JVM通过自旋锁的方式实现的

公平锁和非公平锁

• 公平锁：遵循线程的“先来后到”（即先等待锁的线程先拿到锁）
• 非公平锁：所有线程同等概率抢锁，哪个线程抢到锁，哪个线程先执行

在Java中，synchronized底层就是非公平锁。如果要使用公平锁，可以创建ReentrantLock对象，使用带参的构造方法ReentrantLock(boolean fair)，参数传递true即表示开启公平锁

读写锁

在读者写者问题与读写锁中已经介绍过读写锁，此处不再赘述

在Java标准库中，也提供了读写锁的类：ReentrantReadWriteLock，具体来说：

1. ReentrantReadWriteLock.ReadLock：表示读者锁，提供了对应的加锁lock和解锁unlock方法
2. ReentrantReadWriteLock.WriteLock：表示写者锁，提供了对应的加锁lock和解锁unlock方法

需要注意，synchronized底层不是读写锁

synchronized原理

在Java中，sychronized加锁过程不只是先轻量级锁再重量级锁，具体来说分为下面四个过程：

1. 无锁
2. 偏向锁
3. 轻量级锁
4. 重量级锁

第一次尝试加锁的线程会进入偏向锁，所谓偏向锁并不是真正进行了加锁，而是给线程对象打一个加锁的标记，表示这个锁属于哪一个线程，如果后续没有其他的线程，那么此时就可以节省加锁的开销。当出现了多个线程访问同一个资源时，就会由偏向锁转换为轻量级锁，当锁竞争严重时，轻量级锁就会转换为重量级锁

需要注意的是，上面的过程是不可逆的，也就是说，如果synchronized的锁从偏向锁最后变为重量级锁，那么除非创建新的锁对象，否则该锁之后一直都是重量级锁

除此之外，synchronized还有下面的两种锁优化：

• 锁消除：当JIT编译器通过逃逸分析等手段判断某个synchronized锁对象不会被多个线程共享，或者该同步块不会发生线程竞争时，就可能在编译后消除这部分加锁与解锁操作，从而减少不必要的同步开销
• 锁粗化：当一段连续代码中对同一个锁对象进行了多次相邻的加锁和解锁，并且这些同步区域之间没有可能导致线程安全问题的逻辑时，JIT编译器可能会把多个小的同步块合并为一个更大的同步块，以减少频繁加锁、解锁带来的性能损耗

CAS与原子操作

CAS介绍

见C++并发支持库部分介绍

Java原子操作

在Java中，JDK提供了一组原子操作类，位于java.util.concurrent.atomic包下。这些类可以在不显式加锁的情况下，保证对单个变量的线程安全操作，底层通常依赖CAS（Compare And Set，比较并交换）机制实现。

Java中常见的原子操作类主要分为下面几类：

1. 基本类型原子类（AtomicBoolean、AtomicInteger、AtomicLong）：用于对基本类型进行原子更新，例如自增、自减、赋值、比较并交换等操作
2. 引用类型原子类（AtomicReference<V>）：用于对对象引用进行原子更新，适合在多线程环境下安全地修改某个对象的引用
3. 数组类型原子类（AtomicIntegerArray、AtomicLongArray、AtomicReferenceArray：用于对数组中的某个元素进行原子操作，保证数组元素更新时的线程安全
4. 字段更新器（AtomicIntegerFieldUpdater、AtomicLongFieldUpdater、AtomicReferenceFieldUpdater）：用于对普通对象中的某个volatile字段进行原子更新。相比直接使用原子类，这种方式灵活性更高，但使用起来也更复杂
5. 累加器与加法器（LongAdder、DoubleAdder、LongAccumulator、DoubleAccumulator）：这类原子类更适合高并发场景下的计数和累加操作，其中LongAdder在大量线程同时更新计数器时，通常比AtomicLong具有更好的性能

需要注意，原子类适合对单个共享变量进行线程安全操作，如果涉及多个变量之间的整体一致性，仅使用原子类通常是不够的，仍然可能需要配合锁机制来保证线程安全

不同原子类虽然操作对象不同，但常见方法的设计思路基本一致，以AtomicInteger为例说明常见原子操作方法，这些方法在AtomicLong等原子类中的使用方式也基本类似。下面给出常见使用方法对照表：

方法	作用
get()	获取当前值
set(newValue)	直接设置新值
lazySet(newValue)	延迟设置新值，不要求立刻对其他线程可见
compareAndSet(expect, update)	如果当前值等于期望值，则以原子方式更新为新值
getAndSet(newValue)	以原子方式设置新值，并返回旧值
incrementAndGet()	原子地加1，返回新值
getAndIncrement()	原子地加1，返回旧值
decrementAndGet()	原子地减1，返回新值
getAndDecrement()	原子地减1，返回旧值
addAndGet(delta)	原子地加上指定值，返回新值
getAndAdd(delta)	原子地加上指定值，返回旧值
updateAndGet(function)	使用函数更新当前值，返回新值
getAndUpdate(function)	使用函数更新当前值，返回旧值
accumulateAndGet(x, function)	将当前值和给定值按指定规则计算后更新，返回新值
getAndAccumulate(x, function)	将当前值和给定值按指定规则计算后更新，返回旧值

对于AtomicReference<V>这类引用类型原子类，常见方法与上面基本一致，只不过操作对象从基本类型变成了对象引用，例如：

方法	作用
get()	获取当前引用
set(newValue)	设置新引用
compareAndSet(expect, update)	如果当前引用等于期望引用，则更新为新引用
getAndSet(newValue)	设置新引用，并返回旧引用
updateAndGet(function)	按指定规则更新引用，并返回新引用

对于高并发计数场景中常见的LongAdder，常用方法如下：

方法	作用
increment()	原子地加1
decrement()	原子地减1
add(x)	原子地加上指定值
sum()	获取当前总和
reset()	重置为0
sumThenReset()	获取当前总和后再重置为0

一般来说：

1. 如果只是普通计数器，常用AtomicInteger或AtomicLong
2. 如果是高并发统计计数，更常使用LongAdder
3. 如果需要对对象引用进行原子更新，则使用AtomicReference

下面通过AtomicInteger演示一个线程安全计数器的实现：

import java.util.concurrent.atomic.AtomicInteger;

public class Test {
    public static void main(String[] args) throws InterruptedException {
        AtomicInteger count = new AtomicInteger(0);

        Thread t1 = new Thread(() -> {
            for (int i = 0; i < 1000; i++) {
                count.incrementAndGet();
            }
        });

        Thread t2 = new Thread(() -> {
            for (int i = 0; i < 1000; i++) {
                count.incrementAndGet();
            }
        });

        t1.start();
        t2.start();

        t1.join();
        t2.join();

        System.out.println(count.get());
    }
}

上述代码中，两个线程分别对count执行1000次自增操作，最终结果一定是2000。如果这里使用普通的int变量进行count++操作，在多线程环境下就可能出现线程安全问题

ABA问题

虽然CAS可以保证单次比较并交换操作的原子性，但它只会比较“当前值是否等于期望值”，不会关心这个值在比较期间是否被其他线程改动过。因此就可能出现ABA问题

假设某个共享变量的初始值为A，线程1准备通过CAS将它修改为B。在线程1真正执行CAS之前，线程2先把这个变量从A改成了B，随后又从B改回了A。这时线程1再次执行CAS时，看到当前值仍然是A，就会认为这个变量在此期间没有发生变化，于是成功把它更新为B。但实际上，这个值已经被其他线程改动过，只是最后又回到了原值，这种“值变了又变回去”的情况就是ABA问题

根据上面的过程不难看出，ABA问题的本质不是“CAS失去了原子性”，而是“CAS只比较值本身，无法识别值的变化历史”。在某些场景下，这会导致程序误判共享变量始终未被其他线程修改过

在Java中，常见的解决思路是给数据额外附带一个版本号或标记位，让CAS在比较值的同时也比较版本信息。JDK提供了两种常见原子类：

1. AtomicStampedReference：为引用额外维护一个整数版本号（stamp），每次更新时同时修改引用和值对应的版本号，适合通过“版本递增”的方式解决ABA问题
2. AtomicMarkableReference：为引用额外维护一个布尔标记位（mark），适合只需要标识“是否被修改过”这类场景

版本号方案的核心很简单：不只比较“值”，还要同时比较“这个值是第几版”。比如一个变量初始是A，版本号是1。线程1读取到的是A,1，准备更新成B,2。这时线程2先把它改成B,2，又改回A,3。虽然当前值又变成了A，但版本号已经不是1 了，而是3。于是线程1再执行CAS时，会发现“值看起来对，但版本号不对”，更新就会失败，这样就识别出了ABA

可以，下面这版是整理后的笔记风格，适合直接放到文档里。

信号量Semaphore

Semaphore表示信号量，本质上是一个“许可证计数器”。它内部维护若干个许可证，线程在访问共享资源之前需要先获取许可证，使用完之后再归还许可证

如果当前还有可用许可证，线程就可以继续执行；如果许可证已经用完，线程就需要等待其他线程释放许可证

常见使用场景如下：

1. 控制数据库连接池中同时可用的连接数量
2. 限制某段业务代码的最大并发执行线程数
3. 控制多个线程对有限资源的访问

常用方法如下：

方法	作用
acquire()	获取一个许可证，若没有可用许可证则阻塞等待
release()	释放一个许可证
tryAcquire()	尝试获取许可证，获取成功返回true，否则返回false
tryAcquire(long timeout, TimeUnit unit)	在指定时间内尝试获取许可证
availablePermits()	获取当前剩余的许可证数量

需要注意的是，Semaphore既可以设置为公平模式，也可以设置为非公平模式。公平模式下，等待时间更久的线程会优先获取许可证；非公平模式下，不保证获取顺序，但通常吞吐量更高

例如下面的示例：

import java.util.concurrent.Semaphore;

public class Test {
    public static void main(String[] args) {
        Semaphore semaphore = new Semaphore(2);

        for (int i = 1; i <= 5; i++) {
            int carId = i;
            new Thread(() -> {
                try {
                    System.out.println("车辆" + carId + "准备进入停车场");
                    semaphore.acquire();
                    System.out.println("车辆" + carId + "进入停车场");

                    Thread.sleep(2000);

                    System.out.println("车辆" + carId + "驶离停车场");
                    semaphore.release();
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }).start();
        }
    }
}

倒计时器CountDownLatch

CountDownLatch表示倒计时门闩，它内部维护一个计数器。一个或多个线程可以调用await()（可以理解为all wait）进入等待状态，而其他线程每完成一次指定任务，就调用一次countDown()将计数减一。当计数减为0时，所有等待的线程都会被唤醒并继续执行

常见使用场景如下：

1. 主线程等待多个子线程执行结束
2. 系统启动时等待多个模块初始化完成
3. 将一个大任务拆成多个子任务，等待所有子任务执行完成后再汇总结果

常用方法如下：

方法	作用
await()	当前线程等待，直到计数变为0
await(long timeout, TimeUnit unit)	当前线程在指定时间内等待
countDown()	将计数减1
getCount()	获取当前计数值

需要注意，CountDownLatch是一次性的。计数减到0之后就不能重置，如果需要重复使用，则通常考虑CyclicBarrier

import java.util.concurrent.CountDownLatch;

public class Test {
    public static void main(String[] args) throws InterruptedException {
        CountDownLatch latch = new CountDownLatch(3);

        for (int i = 1; i <= 3; i++) {
            int threadId = i;
            new Thread(() -> {
                System.out.println("线程" + threadId + "开始执行");

                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }

                System.out.println("线程" + threadId + "执行结束");
                latch.countDown();
            }).start();
        }

        System.out.println("主线程等待子线程执行完毕");
        latch.await();
        System.out.println("所有子线程都执行完毕，主线程继续执行");
    }
}

JUC线程安全的集合类

线程安全的ArrayList

要想使用线程安全的ArrayList有下面三种方式：

1. 自行加锁
2. 使用Collections.synchronizedList(new ArrayList)创建一个线程安全的ArrayList
3. 使用写时拷贝CopyOnWriteArrayList，但是这个方案适合多个线程读，一个线程写并且内容较少的情况

线程安全的队列

使用阻塞队列BlockingQueue，其实现类下面几种：

1. ArrayBlockingQueue：基于数组、有界、FIFO。容量创建后固定，适合“固定缓冲区”场景。支持公平锁参数（fair=true 时更公平但吞吐通常更低）
2. LinkedBlockingQueue：基于链表、可选有界（默认接近无界）、FIFO。一般吞吐不错，常用于生产者-消费者队列
3. SynchronousQueue不存储元素（容量为 0），生产者和消费者必须一一“交接”。适合任务直接移交（handoff）场景
4. PriorityBlockingQueue：无界优先级队列，出队按优先级，不保证同优先级元素的先后顺序。适合“优先级任务调度”
5. DelayQueue：无界延迟队列，元素必须实现 Delayed，只有“到期”元素才能被取出。适合定时任务、超时处理

BlockingQueue常用方法可以按照“操作失败时如何处理”来理解，官方接口将其分成下面四种形式：

操作类型	抛出异常	返回特殊值	阻塞等待	超时等待
插入元素	add(e)	offer(e)	put(e)	offer(e, time, unit)
删除队头元素	remove()	poll()	take()	poll(time, unit)
查看队头元素	element()	peek()	不适用	不适用

上表可以理解为：

1. add(e)和remove()更偏向普通Queue风格，失败时直接抛异常
2. offer(e)和poll()失败时不会阻塞，而是返回false或null
3. put(e)和take()是阻塞队列最常用的方法，适合典型的生产者-消费者模型
4. offer(e, time, unit)和poll(time, unit)适合“不想一直等下去”的场景

除此之外，还有两个很常用的辅助方法：

方法	作用
remainingCapacity()	返回当前理论上还能插入多少个元素；如果是无界队列，一般返回Integer.MAX_VALUE
drainTo(collection)	一次性将当前队列中可取出的元素批量转移到另一个集合中

需要注意：

1. BlockingQueue不允许插入null，因为poll()会使用null表示“获取失败”
2. remainingCapacity()只能作为参考，不能据此保证下一次插入一定成功，因为多线程下队列状态随时可能变化

线程安全的哈希表

使用ConcurrentHashMap，使用方式与普通的HashMap一致，下面重点关注ConcurrentHashMap在线程安全问题上的处理：

1. 分桶并发控制：如果桶为空，插入首节点通常通过CAS完成；如果桶不为空，更新操作会锁住该桶（基于桶头节点的synchronized），而不是锁整张表
2. 计数器修改使用CAS
3. 读操作不加锁，但是使用volatile保证内存可见性
4. 扩容采用协作迁移：扩容时不会一次性搬完所有桶，而是把迁移任务分段后由多个线程共同完成；迁移期间通过转发节点引导读写在旧表/新表之间正确进行，降低单次扩容停顿