Java线程同步机制详解
哪吒 2024/1/15
# Java线程同步机制详解
点击勘误issues (opens new window)，哪吒感谢大家的阅读
# 1. 线程同步概述
# 1.1 为什么需要线程同步
在多线程环境中，多个线程可能同时访问共享资源，如果没有适当的同步机制，可能导致：
- 数据不一致：多个线程同时修改同一数据
- 竞态条件：程序结果依赖于线程执行的时序
- 内存可见性问题：一个线程的修改对其他线程不可见
# 1.2 Java内存模型
public class MemoryVisibilityDemo {
private static boolean flag = false;
public static void main(String[] args) throws InterruptedException {
Thread thread1 = new Thread(() -> {
while (!flag) {
// 可能永远不会结束，因为flag的修改对当前线程不可见
}
System.out.println("线程1结束");
});
Thread thread2 = new Thread(() -> {
try {
Thread.sleep(1000);
flag = true;
System.out.println("flag设置为true");
} catch (InterruptedException e) {
e.printStackTrace();
}
});
thread1.start();
thread2.start();
thread1.join();
thread2.join();
}
}
# 2. synchronized关键字
# 2.1 synchronized方法
public class SynchronizedMethodDemo {
private int count = 0;
// 同步实例方法
public synchronized void increment() {
count++;
}
// 同步静态方法
public static synchronized void staticMethod() {
System.out.println("静态同步方法");
}
public synchronized int getCount() {
return count;
}
public static void main(String[] args) throws InterruptedException {
SynchronizedMethodDemo demo = new SynchronizedMethodDemo();
Thread[] threads = new Thread[10];
for (int i = 0; i < 10; i++) {
threads[i] = new Thread(() -> {
for (int j = 0; j < 1000; j++) {
demo.increment();
}
});
}
for (Thread thread : threads) {
thread.start();
}
for (Thread thread : threads) {
thread.join();
}
System.out.println("最终计数: " + demo.getCount()); // 10000
}
}
# 2.2 synchronized代码块
public class SynchronizedBlockDemo {
private int count1 = 0;
private int count2 = 0;
private final Object lock1 = new Object();
private final Object lock2 = new Object();
public void method1() {
synchronized (lock1) {
count1++;
System.out.println("Method1: " + count1);
}
}
public void method2() {
synchronized (lock2) {
count2++;
System.out.println("Method2: " + count2);
}
}
// 细粒度锁，提高并发性能
public void finegrainedLocking() {
// 只同步必要的代码段
int localVar = 0;
// 一些不需要同步的操作
localVar++;
synchronized (lock1) {
// 只有这部分需要同步
count1 += localVar;
}
}
}
# 2.3 synchronized的实现原理
public class SynchronizedPrincipleDemo {
private final Object monitor = new Object();
public void demonstrateMonitor() {
synchronized (monitor) {
// 进入同步块时获取monitor锁
System.out.println("在同步块中执行");
// 退出同步块时释放monitor锁
}
}
// 等价于上面的代码（概念上）
public void equivalentCode() {
// monitorenter
try {
System.out.println("在同步块中执行");
} finally {
// monitorexit
}
}
}
# 3. volatile关键字
# 3.1 volatile的作用
public class VolatileDemo {
private volatile boolean flag = false;
private volatile int count = 0;
public void writer() {
count = 42;
flag = true; // volatile写，确保之前的写操作对其他线程可见
}
public void reader() {
if (flag) { // volatile读，确保能看到最新值
System.out.println("Count: " + count); // 保证能看到42
}
}
public static void main(String[] args) {
VolatileDemo demo = new VolatileDemo();
Thread writerThread = new Thread(demo::writer);
Thread readerThread = new Thread(() -> {
while (!demo.flag) {
// 等待flag变为true
}
demo.reader();
});
readerThread.start();
writerThread.start();
}
}
# 3.2 volatile vs synchronized
public class VolatileVsSynchronized {
private volatile int volatileCount = 0;
private int synchronizedCount = 0;
// volatile不能保证原子性
public void incrementVolatile() {
volatileCount++; // 不是原子操作！
}
// synchronized保证原子性
public synchronized void incrementSynchronized() {
synchronizedCount++; // 原子操作
}
// 正确使用volatile的场景
private volatile boolean shutdown = false;
public void doWork() {
while (!shutdown) {
// 执行工作
}
}
public void shutdown() {
shutdown = true; // 简单的标志位设置
}
}
# 4. wait()和notify()机制
# 4.1 基本用法
public class WaitNotifyDemo {
private final Object lock = new Object();
private boolean condition = false;
public void waitingMethod() {
synchronized (lock) {
while (!condition) {
try {
System.out.println("等待条件满足...");
lock.wait(); // 释放锁并等待
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return;
}
}
System.out.println("条件满足，继续执行");
}
}
public void notifyingMethod() {
synchronized (lock) {
condition = true;
System.out.println("条件已设置，通知等待线程");
lock.notify(); // 唤醒一个等待线程
// lock.notifyAll(); // 唤醒所有等待线程
}
}
public static void main(String[] args) throws InterruptedException {
WaitNotifyDemo demo = new WaitNotifyDemo();
Thread waitingThread = new Thread(demo::waitingMethod);
Thread notifyingThread = new Thread(demo::notifyingMethod);
waitingThread.start();
Thread.sleep(1000); // 确保等待线程先启动
notifyingThread.start();
waitingThread.join();
notifyingThread.join();
}
}
# 4.2 生产者消费者模式
import java.util.LinkedList;
import java.util.Queue;
public class ProducerConsumerDemo {
private final Queue<Integer> queue = new LinkedList<>();
private final int capacity = 5;
private final Object lock = new Object();
public void produce() throws InterruptedException {
int value = 0;
while (true) {
synchronized (lock) {
while (queue.size() == capacity) {
System.out.println("队列已满，生产者等待...");
lock.wait();
}
queue.offer(value);
System.out.println("生产: " + value);
value++;
lock.notifyAll(); // 通知消费者
}
Thread.sleep(1000);
}
}
public void consume() throws InterruptedException {
while (true) {
synchronized (lock) {
while (queue.isEmpty()) {
System.out.println("队列为空，消费者等待...");
lock.wait();
}
int value = queue.poll();
System.out.println("消费: " + value);
lock.notifyAll(); // 通知生产者
}
Thread.sleep(1500);
}
}
public static void main(String[] args) {
ProducerConsumerDemo demo = new ProducerConsumerDemo();
Thread producer = new Thread(() -> {
try {
demo.produce();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
Thread consumer = new Thread(() -> {
try {
demo.consume();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
producer.start();
consumer.start();
}
}
# 5. Lock接口
# 5.1 ReentrantLock基本用法
import java.util.concurrent.locks.ReentrantLock;
public class ReentrantLockDemo {
private final ReentrantLock lock = new ReentrantLock();
private int count = 0;
public void increment() {
lock.lock();
try {
count++;
System.out.println("Count: " + count);
} finally {
lock.unlock(); // 必须在finally中释放锁
}
}
public boolean tryIncrement() {
if (lock.tryLock()) {
try {
count++;
System.out.println("Try increment success: " + count);
return true;
} finally {
lock.unlock();
}
} else {
System.out.println("无法获取锁，跳过操作");
return false;
}
}
public static void main(String[] args) {
ReentrantLockDemo demo = new ReentrantLockDemo();
// 创建多个线程测试
for (int i = 0; i < 3; i++) {
new Thread(() -> {
for (int j = 0; j < 3; j++) {
demo.increment();
demo.tryIncrement();
}
}).start();
}
}
}
# 5.2 ReadWriteLock读写锁
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
public class ReadWriteLockDemo {
private final ReadWriteLock lock = new ReentrantReadWriteLock();
private String data = "初始数据";
public String read() {
lock.readLock().lock();
try {
System.out.println(Thread.currentThread().getName() + " 正在读取: " + data);
Thread.sleep(1000); // 模拟读取耗时
return data;
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return null;
} finally {
lock.readLock().unlock();
}
}
public void write(String newData) {
lock.writeLock().lock();
try {
System.out.println(Thread.currentThread().getName() + " 正在写入: " + newData);
Thread.sleep(1000); // 模拟写入耗时
this.data = newData;
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
lock.writeLock().unlock();
}
}
public static void main(String[] args) {
ReadWriteLockDemo demo = new ReadWriteLockDemo();
// 创建多个读线程
for (int i = 0; i < 3; i++) {
new Thread(() -> demo.read(), "读线程-" + i).start();
}
// 创建写线程
new Thread(() -> demo.write("新数据"), "写线程").start();
// 再创建读线程
for (int i = 3; i < 6; i++) {
new Thread(() -> demo.read(), "读线程-" + i).start();
}
}
}
# 6. 条件变量Condition
# 6.1 Condition基本用法
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
public class ConditionDemo {
private final ReentrantLock lock = new ReentrantLock();
private final Condition condition = lock.newCondition();
private boolean ready = false;
public void waitForCondition() {
lock.lock();
try {
while (!ready) {
System.out.println("等待条件满足...");
condition.await(); // 等待条件
}
System.out.println("条件满足，继续执行");
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
lock.unlock();
}
}
public void signalCondition() {
lock.lock();
try {
ready = true;
System.out.println("条件已满足，发送信号");
condition.signal(); // 唤醒等待线程
} finally {
lock.unlock();
}
}
public static void main(String[] args) throws InterruptedException {
ConditionDemo demo = new ConditionDemo();
Thread waitingThread = new Thread(demo::waitForCondition);
Thread signalingThread = new Thread(demo::signalCondition);
waitingThread.start();
Thread.sleep(1000);
signalingThread.start();
waitingThread.join();
signalingThread.join();
}
}
# 6.2 多条件变量示例
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
public class MultiConditionDemo {
private final ReentrantLock lock = new ReentrantLock();
private final Condition notFull = lock.newCondition();
private final Condition notEmpty = lock.newCondition();
private final Object[] items = new Object[10];
private int putIndex = 0;
private int takeIndex = 0;
private int count = 0;
public void put(Object item) throws InterruptedException {
lock.lock();
try {
while (count == items.length) {
notFull.await(); // 等待不满条件
}
items[putIndex] = item;
putIndex = (putIndex + 1) % items.length;
count++;
System.out.println("放入元素: " + item + ", 当前数量: " + count);
notEmpty.signal(); // 通知不空条件
} finally {
lock.unlock();
}
}
public Object take() throws InterruptedException {
lock.lock();
try {
while (count == 0) {
notEmpty.await(); // 等待不空条件
}
Object item = items[takeIndex];
items[takeIndex] = null;
takeIndex = (takeIndex + 1) % items.length;
count--;
System.out.println("取出元素: " + item + ", 当前数量: " + count);
notFull.signal(); // 通知不满条件
return item;
} finally {
lock.unlock();
}
}
}
# 7. 同步机制比较
# 7.1 性能比较
import java.util.concurrent.locks.ReentrantLock;
public class SynchronizationComparison {
private int synchronizedCount = 0;
private int lockCount = 0;
private final Object syncLock = new Object();
private final ReentrantLock reentrantLock = new ReentrantLock();
public void synchronizedIncrement() {
synchronized (syncLock) {
synchronizedCount++;
}
}
public void lockIncrement() {
reentrantLock.lock();
try {
lockCount++;
} finally {
reentrantLock.unlock();
}
}
public static void performanceTest() {
SynchronizationComparison demo = new SynchronizationComparison();
int threadCount = 10;
int iterations = 100000;
// 测试synchronized性能
long startTime = System.currentTimeMillis();
Thread[] syncThreads = new Thread[threadCount];
for (int i = 0; i < threadCount; i++) {
syncThreads[i] = new Thread(() -> {
for (int j = 0; j < iterations; j++) {
demo.synchronizedIncrement();
}
});
}
for (Thread thread : syncThreads) {
thread.start();
}
for (Thread thread : syncThreads) {
try {
thread.join();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
long syncTime = System.currentTimeMillis() - startTime;
System.out.println("Synchronized耗时: " + syncTime + "ms");
// 测试ReentrantLock性能
startTime = System.currentTimeMillis();
Thread[] lockThreads = new Thread[threadCount];
for (int i = 0; i < threadCount; i++) {
lockThreads[i] = new Thread(() -> {
for (int j = 0; j < iterations; j++) {
demo.lockIncrement();
}
});
}
for (Thread thread : lockThreads) {
thread.start();
}
for (Thread thread : lockThreads) {
try {
thread.join();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
long lockTime = System.currentTimeMillis() - startTime;
System.out.println("ReentrantLock耗时: " + lockTime + "ms");
}
public static void main(String[] args) {
performanceTest();
}
}
# 8. 最佳实践
# 8.1 避免死锁
public class DeadlockAvoidance {
private final Object lock1 = new Object();
private final Object lock2 = new Object();
// 错误的做法 - 可能导致死锁
public void badMethod1() {
synchronized (lock1) {
synchronized (lock2) {
System.out.println("Method1执行");
}
}
}
public void badMethod2() {
synchronized (lock2) {
synchronized (lock1) {
System.out.println("Method2执行");
}
}
}
// 正确的做法 - 按顺序获取锁
public void goodMethod1() {
synchronized (lock1) {
synchronized (lock2) {
System.out.println("GoodMethod1执行");
}
}
}
public void goodMethod2() {
synchronized (lock1) { // 同样的顺序
synchronized (lock2) {
System.out.println("GoodMethod2执行");
}
}
}
}
# 8.2 减少锁的粒度
import java.util.concurrent.ConcurrentHashMap;
public class LockGranularity {
// 粗粒度锁 - 整个对象
private final Object coarseLock = new Object();
private int value1 = 0;
private int value2 = 0;
public void coarseGrainedMethod() {
synchronized (coarseLock) {
value1++;
value2++;
}
}
// 细粒度锁 - 分别保护不同的资源
private final Object lock1 = new Object();
private final Object lock2 = new Object();
public void fineGrainedMethod1() {
synchronized (lock1) {
value1++;
}
}
public void fineGrainedMethod2() {
synchronized (lock2) {
value2++;
}
}
// 使用并发容器
private final ConcurrentHashMap<String, Integer> concurrentMap = new ConcurrentHashMap<>();
public void useConcurrentCollection() {
concurrentMap.put("key1", 1);
concurrentMap.put("key2", 2);
// 不需要显式同步
}
}
# 9. 总结
本文详细介绍了Java线程同步机制，包括：
- synchronized关键字：方法同步和代码块同步
- volatile关键字：保证内存可见性
- wait/notify机制：线程间通信
- Lock接口：更灵活的锁机制
- Condition条件变量：精确的线程控制
- 最佳实践：避免死锁、减少锁粒度
选择合适的同步机制对于编写高效、安全的多线程程序至关重要。在实际开发中，应该根据具体场景选择最适合的同步方式。