并发工具与线程安全

学习目标

从不变性、线程隔离、消息传递和同步四条路径判断线程安全。
了解 synchronized、Lock、原子类、并发集合和同步器的边界。
识别竞态、死锁、活锁、饥饿和阻塞队列堆积等问题。

核心知识点

专业术语

线程安全（thread-safe）：多个线程并发使用时仍满足对象对外约定的不变式。
竞态条件（race condition）：结果依赖线程执行时序，导致同样输入可能得到错误结果。
互斥（mutual exclusion）：同一时刻只允许一个线程进入受保护的临界区。
原子操作（atomic operation）：从其他线程观察时不可被拆开的单步状态变化。
同步器（synchronizer）：CountDownLatch、Semaphore、CyclicBarrier 等协调线程阶段的工具。

白话解释与边界

先尝试让数据不可变或每个线程独享；必须共享时，再选择互斥锁、CAS 原子类、并发集合或消息队列。ConcurrentHashMap 只保证 Map 自己的并发操作安全，业务上的“检查再更新”仍要用 compute、putIfAbsent 或外部同步形成原子边界。

同步工具解决的是不同问题：锁保护共享状态，原子类更新小变量，阻塞队列传递任务，CountDownLatch 等待阶段完成，Semaphore 限制并发名额。不要把所有问题都粗暴地包在一个大锁里。

常用用法

synchronized：保护临界区

实例 synchronized 方法锁住当前对象；静态 synchronized 方法锁住 Class 对象。

java

// 说明：increment() 和 value() 都持有 counter 的监视器，因此 count++ 与后续读取不会被其他线程交叉。
public class SynchronizedCounterDemo {
    private int count;

    synchronized void increment() {
        count++;
    }

    synchronized int value() {
        return count;
    }

    public static void main(String[] args) {
        SynchronizedCounterDemo counter = new SynchronizedCounterDemo();
// 初始状态：counter = new SynchronizedCounterDemo()。
// 作用：SynchronizedCounterDemo counter = new SynchronizedCounterDemo();；保护临界区。
        counter.increment();
// 作用：counter.increment();；保护临界区。
        counter.increment();
// 作用：counter.increment();；保护临界区。
        System.out.println(counter.value());
// 输出：2
    }
}

临界区应短小，不能因为“线程安全”就把网络、磁盘或数据库调用一起锁住。

ReentrantLock：显式获得与释放锁

ReentrantLock 支持可中断、超时、公平策略和多个 Condition；但只要 lock() 成功，就必须在 finally 中释放。

java

// 说明：ReentrantLock：显式获得与释放锁。
import java.util.concurrent.locks.ReentrantLock;

public class ReentrantLockDemo {
    public static void main(String[] args) {
        ReentrantLock lock = new ReentrantLock();
// 初始状态：lock = new ReentrantLock()。
// 作用：ReentrantLock lock = new ReentrantLock();；显式获得与释放锁。
        lock.lock();
// 作用：lock.lock();；显式获得与释放锁。
        try {
            System.out.println("protected");
// 输出：protected
        } finally {
            lock.unlock();
// 作用：lock.unlock();；显式获得与释放锁。
        }
    }
}

详细的 tryLock 和 Condition 见互斥锁专题页。

AtomicInteger：无锁地更新单个计数

原子类适合独立变量的 CAS 更新；多个字段必须一起满足不变式时，单独给每个字段套 Atomic 并不能形成事务边界，应使用锁或不可变快照。

java

// 说明：AtomicInteger：无锁地更新单个计数。
import java.util.concurrent.atomic.AtomicInteger;

public class AtomicIntegerDemo {
    public static void main(String[] args) {
        AtomicInteger count = new AtomicInteger(10);
// 初始状态：count = new AtomicInteger(10)。
// 作用：AtomicInteger count = new AtomicInteger(10);；无锁地更新单个计数。
        int old = count.getAndIncrement();
// 作用：int old = count.getAndIncrement();；无锁地更新单个计数。
// 初始状态：old = count.getAndIncrement()。
        boolean changed = count.compareAndSet(11, 12);
// 作用：boolean changed = count.compareAndSet(11, 12);；无锁地更新单个计数。
// 初始状态：changed = count.compareAndSet(11, 12)。
        System.out.println(old + ", " + changed + ", " + count.get());
// 输出：10, true, 12
    }
}

LongAdder：高竞争统计

LongAdder 通过分散热点降低竞争，适合吞吐统计；sum() 不是一个把所有并发更新冻结的快照，金融余额等精确状态不要用它替代锁或 AtomicLong。

java

// 说明：LongAdder：高竞争统计。
import java.util.concurrent.atomic.LongAdder;

public class LongAdderDemo {
    public static void main(String[] args) {
        LongAdder requests = new LongAdder();
// 初始状态：requests = new LongAdder()。
// 作用：LongAdder requests = new LongAdder();；高竞争统计。
        requests.increment();
// 作用：requests.increment();；高竞争统计。
        requests.add(4);
// 作用：requests.add(4);；高竞争统计。
        System.out.println(requests.sum());
// 输出：5
    }
}

ConcurrentHashMap.computeIfAbsent：并发缓存初始化

computeIfAbsent 用于把“没有值再计算并放入”表达为一个并发操作。

java

// 说明：ConcurrentHashMap.computeIfAbsent：并发缓存初始化。
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

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

计算函数应短小、无递归修改同一个 Map，且不要在其中执行长时间阻塞 I/O。

BlockingQueue：用消息传递代替共享列表

BlockingQueue 用于让生产者只放入任务、消费者只取出任务，并以队列容量形成背压边界。

java

// 说明：BlockingQueue：用消息传递代替共享列表。
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public class BlockingQueueDemo {
    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
    }
}

put/take 会等待，offer/poll 可以超时或立即返回，选择时要明确满队列和空队列的策略。

CountDownLatch：等待一组一次性事件

CountDownLatch 只能从计数值递减到 0 一次，适合“等所有初始化完成”；需要反复使用的阶段协调应考虑 CyclicBarrier 或 Phaser。

java

// 说明：CountDownLatch：等待一组一次性事件。
import java.util.concurrent.CountDownLatch;

public class CountDownLatchDemo {
    public static void main(String[] args) throws InterruptedException {
        CountDownLatch ready = new CountDownLatch(2);
// 初始状态：ready = new CountDownLatch(2)。
// 作用：CountDownLatch ready = new CountDownLatch(2);；等待一组一次性事件。
        ready.countDown();
// 初始状态：ready.countDown()。
// 作用：ready.countDown();；等待一组一次性事件。
        ready.countDown();
// 初始状态：ready.countDown()。
// 作用：ready.countDown();；等待一组一次性事件。
        ready.await();
// 初始状态：ready.await()。
// 作用：ready.await();；等待一组一次性事件。
        System.out.println("all ready");
// 输出：all ready
    }
}

不常用但需要知道

`ThreadPoolExecutor.CallerRunsPolicy`：由提交线程执行拒绝任务

队列已满且线程池仍运行时，它让提交任务的线程同步执行任务，形成自然背压。

java

// 作用：通过 ThreadPoolExecutor.CallerRunsPolicy 由提交线程执行拒绝任务。
import java.util.concurrent.ThreadPoolExecutor;
var policy = new ThreadPoolExecutor.CallerRunsPolicy();
// 初始状态：policy = new ThreadPoolExecutor.CallerRunsPolicy()。
// 作用：var policy = new ThreadPoolExecutor.CallerRunsPolicy();；由提交线程执行拒绝任务。
System.out.println(policy.getClass().getSimpleName());
// 输出：CallerRunsPolicy
System.out.println(policy instanceof ThreadPoolExecutor.CallerRunsPolicy);
// 输出：true

ReadWriteLock：读多写少时分离锁

读锁允许多个读者并行，写锁排斥读写双方；读写比例、临界区长度和升级/降级策略都要用基准验证。若读操作很短，普通 synchronized 可能更简单。

java

// 说明：readLock 持有期间允许其他读者进入，但会阻塞 writeLock；finally 确保当前线程归还读锁。
import java.util.concurrent.locks.ReentrantReadWriteLock;

public class ReadWriteLockDemo {
    public static void main(String[] args) {
        var lock = new ReentrantReadWriteLock();
// 初始状态：lock = new ReentrantReadWriteLock()。
// 作用：var lock = new ReentrantReadWriteLock();；读多写少时分离锁。
        lock.readLock().lock();
// 作用：lock.readLock().lock();；读多写少时分离锁。
        try {
            System.out.println("read snapshot");
// 输出：read snapshot
        } finally {
            lock.readLock().unlock();
// 作用：lock.readLock().unlock();；读多写少时分离锁。
        }
    }
}

StampedLock：乐观读的低频工具

乐观读不真正占用读锁，验证失败后必须回退到读锁。StampedLock 不可重入，也没有 ReentrantLock 那样直观的条件队列；除非读竞争收益明确，否则不要为了“高级”而选它。

java

// 说明：stamp 记录乐观读版本；validate 失败表示期间有写入，必须改持 readLock 重新取 snapshot。
import java.util.concurrent.locks.StampedLock;

public class StampedLockDemo {
    public static void main(String[] args) {
        var lock = new StampedLock();
// 初始状态：lock = new StampedLock()。
// 作用：var lock = new StampedLock();；乐观读的低频工具。
        long stamp = lock.tryOptimisticRead();
// 作用：long stamp = lock.tryOptimisticRead();；乐观读的低频工具。
// 初始状态：stamp = lock.tryOptimisticRead()。
        int snapshot = 42;
// 初始状态：snapshot = 42。
        if (!lock.validate(stamp)) {
            stamp = lock.readLock();
// 作用：stamp = lock.readLock();；乐观读的低频工具，返回读取结果。
// 初始状态：stamp = lock.readLock()。
            try {
                snapshot = 42;
// 初始状态：snapshot = 42。
            } finally {
                lock.unlockRead(stamp);
// 作用：lock.unlockRead(stamp);；乐观读的低频工具。
            }
        }
        System.out.println(snapshot);
// 输出：42
    }
}

Phaser：可变参与者的多阶段协调

Phaser 可以动态注册参与者并推进多个 phase；参与者固定且只有一轮等待时，CountDownLatch 更容易读懂。

java

// 说明：new Phaser(1) 把 main 注册为唯一参与者；arriveAndDeregister() 使未到达数归零并终止 phaser。
import java.util.concurrent.Phaser;

public class PhaserDemo {
    public static void main(String[] args) {
        Phaser phaser = new Phaser(1);
// 初始状态：phaser = new Phaser(1)。
// 作用：Phaser phaser = new Phaser(1);；可变参与者的多阶段协调。
        phaser.arriveAndDeregister();
// 作用：phaser.arriveAndDeregister();；可变参与者的多阶段协调。
        System.out.println(phaser.isTerminated());
// 输出：true
    }
}

ConcurrentLinkedQueue：非阻塞队列

它适合短暂的非阻塞并发收发，不会因为空队列等待，也不提供容量背压；需要阻塞或限制堆积时使用 BlockingQueue。

java

// 说明：offer("a") 由当前线程无阻塞入队，poll() 原子地移除队头；空队列时会立即返回 null。
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
    }
}

简单案例

java

import java.util.concurrent.atomic.AtomicInteger;

public class ThreadSafetyChoiceDemo {
    public static void main(String[] args) {
        AtomicInteger success = new AtomicInteger();
// 初始状态：success = new AtomicInteger()。
// 作用：AtomicInteger success = new AtomicInteger();；非阻塞队列。
        for (int i = 0; i < 3; i++) {
// 初始状态：i = 0; i < 3; i++)。
            success.incrementAndGet();
// 作用：success.incrementAndGet();；非阻塞队列。
        }
        System.out.println("success=" + success.get());
// 输出：success=3
    }
}

先确认共享状态是什么，再决定使用 Atomic、锁、并发集合还是消息传递；工具本身不是线程安全设计的替代品。

易混点

volatile 只提供可见性和有序性，不能把 count++ 变成原子操作。
ConcurrentHashMap 的单次方法安全不等于多步业务流程安全。
LongAdder 适合统计吞吐，不保证每次读取都是全局精确瞬时快照。
CountDownLatch 是一次性的；CyclicBarrier/Phaser 适合重复阶段协调。
ReentrantLock 不是“更安全”的 synchronized；它只提供更多控制，同时增加了手动释放的责任。

课后小问

什么时候优先用 AtomicInteger 而不是 synchronized？答案：共享状态是一个独立变量、更新可以用 CAS 表达且不需要同时保护其他字段时。解析：如果多个字段需要保持一致，仍应使用锁、不可变对象或更高层的原子操作。
为什么 ConcurrentLinkedQueue 不能直接替代 BlockingQueue？答案：它不会等待，也没有容量边界，空取和无限生产都需要调用方自行处理。解析：生产者/消费者需要背压时应使用有界 BlockingQueue 的 put/offer 策略。
为什么锁住整个网络调用通常不是好主意？答案：I/O 等待会让锁长期被占用，其他线程无法进入临界区，吞吐下降甚至造成级联阻塞。解析：锁内只更新共享状态，外部调用放在锁外，并通过队列或状态机衔接。

本节小结

线程安全可以来自不变性、隔离、消息传递或正确同步。
锁保护不变式，原子类适合独立变量，并发集合和阻塞队列分别解决容器并发与任务传递。
同步器要按“一次等待、重复阶段或并发名额”等语义选择。
工具选对只是起点，还要限定临界区、容量、取消和异常边界。

快速回顾

能把竞态、锁、CAS、阻塞队列和同步器区分开。
能判断单变量更新是否适合 Atomic 或 LongAdder。
能解释 ConcurrentHashMap 为什么不能自动保证复合业务原子性。
能根据等待阶段选择 CountDownLatch、CyclicBarrier、Phaser 或 Semaphore。

并发工具与线程安全 ​

学习目标 ​

核心知识点 ​

专业术语 ​

白话解释与边界 ​

常用用法 ​

synchronized：保护临界区 ​

ReentrantLock：显式获得与释放锁 ​

AtomicInteger：无锁地更新单个计数 ​

LongAdder：高竞争统计 ​

ConcurrentHashMap.computeIfAbsent：并发缓存初始化 ​

BlockingQueue：用消息传递代替共享列表 ​

CountDownLatch：等待一组一次性事件 ​

不常用但需要知道 ​

ThreadPoolExecutor.CallerRunsPolicy：由提交线程执行拒绝任务 ​

ReadWriteLock：读多写少时分离锁 ​

StampedLock：乐观读的低频工具 ​

Phaser：可变参与者的多阶段协调 ​

ConcurrentLinkedQueue：非阻塞队列 ​

简单案例 ​

易混点 ​

课后小问 ​

本节小结 ​

快速回顾 ​

并发工具与线程安全

学习目标

核心知识点

专业术语

白话解释与边界

常用用法

synchronized：保护临界区

ReentrantLock：显式获得与释放锁

AtomicInteger：无锁地更新单个计数

LongAdder：高竞争统计

ConcurrentHashMap.computeIfAbsent：并发缓存初始化

BlockingQueue：用消息传递代替共享列表

CountDownLatch：等待一组一次性事件

不常用但需要知道

`ThreadPoolExecutor.CallerRunsPolicy`：由提交线程执行拒绝任务

ReadWriteLock：读多写少时分离锁

StampedLock：乐观读的低频工具

Phaser：可变参与者的多阶段协调

ConcurrentLinkedQueue：非阻塞队列

简单案例

易混点

课后小问

本节小结

快速回顾