多个线程同时写一个Map崩了——Java并发编程核心

// synchronized的三种使用方式
 
// 1. 同步方法（锁的是this对象）
public synchronized void addEvent(Event e) {
    events.add(e);
}
 
// 2. 同步代码块（锁的是指定对象，更细粒度）
public void addEvent(Event e) {
    synchronized (events) {  // 只锁events集合
        events.add(e);
    }
}
 
// 3. 静态同步方法（锁的是Class对象）
public static synchronized void addGlobal(Event e) {
    globalEvents.add(e);
}
 
// JDK 6+的synchronized优化（锁升级）：
// 无锁 ==> 偏向锁（单线程场景，几乎零开销）
//      ==> 轻量级锁（短暂竞争，CAS自旋）
//      ==> 重量级锁（长时间竞争，OS互斥量）
// 大部分场景不会升级到重量级锁

// ReentrantLock vs synchronized
private final ReentrantLock lock = new ReentrantLock();
 
public void addEvent(Event e) {
    lock.lock();
    try {
        events.add(e);
    } finally {
        lock.unlock();  // 必须在finally里释放！
    }
}
 
// ReentrantLock的独有功能：
// 1. tryLock()：尝试获取锁，获取不到立即返回（不阻塞）
if (lock.tryLock(100, TimeUnit.MILLISECONDS)) {
    try { /* 处理 */ } finally { lock.unlock(); }
} else { /* 获取锁超时的处理 */ }
 
// 2. 公平锁：new ReentrantLock(true)，先等先得
// 3. 可中断：lockInterruptibly()，等锁时可以被interrupt
// 4. 多条件：lock.newCondition()，实现精细的等待/通知

volatile——轻量级的可见性保证

// volatile保证可见性，但不保证原子性
private volatile boolean isRunning = true;
 
// 线程A：
isRunning = false;  // 写入后立即刷新到主内存
 
// 线程B：
while (isRunning) {  // 每次读都从主内存读最新值
    // 没有volatile的话，线程B可能一直看到true（CPU缓存）
}
 
// volatile的典型应用：状态标志位
// Flink的TaskManager用volatile标志来控制Task的停止
 
// volatile不保证原子性！
private volatile int count = 0;
count++;  // 不是线程安全的！
// count++ = 读取count + 加1 + 写回count
// 两个线程同时读到0，都加1写回1，结果是1而不是2
// 要用AtomicInteger代替

CAS（Compare And Swap）——无锁的原子操作

// CAS是CPU级别的原子操作
// 比较内存值和预期值，相等则更新为新值
 
// AtomicInteger底层就是CAS
AtomicInteger counter = new AtomicInteger(0);
counter.incrementAndGet();  // 线程安全的i++
 
// CAS的伪代码：
// boolean CAS(address, expectedValue, newValue) {
//     if (memory[address] == expectedValue) {
//         memory[address] = newValue;
//         return true;  // 成功
//     }
//     return false;  // 失败，其他线程已经改了
// }
 
// CAS的问题：ABA问题
// 值从A变成B再变回A，CAS认为没变过
// 解决：AtomicStampedReference（带版本号的CAS）
 
// CAS在大数据中的应用：
// ConcurrentHashMap的put用CAS（桶为空时）
// Flink的MailboxExecutor用CAS操作队列
// Kafka的offset提交用CAS保证原子性

大数据场景的并发编程最佳实践

场景	推荐方案	不推荐	原因
共享Map读写	ConcurrentHashMap	synchronized + HashMap	ConcurrentHashMap锁粒度更细
共享队列	ConcurrentLinkedQueue	synchronized + ArrayList	无锁队列性能更好
计数器/指标	AtomicLong / LongAdder	synchronized + long	CAS比锁轻量
状态标志位	volatile boolean	synchronized	volatile够用且更轻
Flink算子内状态	Flink State API	自定义锁+集合	State天然线程安全+可Checkpoint

实战踩坑：Flink Async IO的并发问题

在某元宇宙社交公司用Flink Async IO查询Redis做维表JOIN。异步回调在IO线程池中执行，需要把Redis查询结果收集起来再发送到下游。

一开始用ArrayList：异步回调的多个线程同时往ArrayList里add，偶尔出现ArrayIndexOutOfBoundsException（ArrayList扩容不是线程安全的）。

改成synchronized + ArrayList：线程安全了但性能下降30%——每次add都要竞争锁。

最终方案ConcurrentLinkedQueue：无锁的线程安全队列，add操作用CAS实现，不需要加锁。性能恢复到接近无锁ArrayList的水平，且线程安全。

教训：大数据代码里遇到多线程场景，优先选JDK的并发集合（ConcurrentXxx），而不是手动加synchronized。并发集合的锁粒度和优化远比手写的锁好。

面试高频追问

追问一：synchronized和ReentrantLock怎么选？

大部分场景用synchronized就够了——JDK 6之后synchronized有偏向锁/轻量级锁优化，性能和ReentrantLock差距很小。只有需要tryLock（超时机制）、公平锁、多条件等高级功能时才用ReentrantLock。Flink和Kafka的源码里两者都有使用。

追问二：LongAdder和AtomicLong有什么区别？

高并发下LongAdder比AtomicLong快很多。AtomicLong所有线程都CAS同一个变量，竞争激烈。LongAdder把计数分散到多个Cell中（类似ConcurrentHashMap的分段思想），减少了CAS竞争。最后sum()时把所有Cell加起来。Flink的Metrics和Kafka的指标计数器都用LongAdder。

追问三：死锁怎么排查？

用jstack 打印线程栈，看有没有线程互相等待对方持有的锁。jstack会自动检测死锁并输出'Found one Java-level deadlock'。避免死锁的最佳实践：所有线程按相同的顺序获取锁、使用tryLock设置超时、尽量缩小锁的范围。

总结

并发编程面试讲清楚四个工具：① synchronized（最简单，JDK 6+有锁升级优化）；② ReentrantLock（更灵活，支持tryLock/公平锁/多条件）；③ volatile（轻量级可见性保证，不保证原子性）；④ CAS（无锁原子操作，AtomicXxx/LongAdder）。大数据场景的最佳实践是用并发集合替代手动加锁。能讲出Flink Async IO的并发踩坑和从ArrayList到ConcurrentLinkedQueue的优化过程，面试官会觉得你在实际项目中处理过并发问题。

下一篇聊线程池：Spark Executor内部有多少线程？Flink的线程模型是什么样的？线程池怎么配最合理？

你们写大数据代码遇到过并发问题吗？怎么解决的？评论区聊聊～

关注「Wood聊大数据」| 9年大数据老兵，聊技术、聊职场、聊成长