学习目标

  • 理解什么是线程安全以及为什么它在多线程编程中至关重要
  • 掌握共享资源访问时产生竞态条件的原因和表现
  • 学习识别典型的线程安全问题和并发风险
  • 通过实例学习如何解决多线程计数器问题

1. 什么是线程安全

1.1 线程安全的定义

线程安全是指在多线程环境下，程序能够正确地处理共享资源，确保数据的一致性和正确性。一个线程安全的程序，无论有多少线程并发执行，都不会出现数据破坏或者得到错误结果的情况。

  1. package org.devlive.tutorial.multithreading.chapter04;
  3. /**
  4.  * 线程安全与非线程安全的对比
  5.  */
  6. public class ThreadSafetyDemo {
  7.     public static void main(String[] args) throws InterruptedException {
  8.         // 演示线程不安全的计数器
  9.         UnsafeCounter unsafeCounter = new UnsafeCounter();
  10.         runCounterTest("线程不安全的计数器", unsafeCounter);
  12.         // 演示线程安全的计数器
  13.         SafeCounter safeCounter = new SafeCounter();
  14.         runCounterTest("线程安全的计数器", safeCounter);
  15.     }
  17.     // 线程不安全的计数器
  18.     static class UnsafeCounter {
  19.         private int count = 0;
  21.         public void increment() {
  22.             count++;  // 非原子操作
  23.         }
  25.         public int getCount() {
  26.             return count;
  27.         }
  28.     }
  30.     // 线程安全的计数器
  31.     static class SafeCounter {
  32.         private int count = 0;
  34.         public synchronized void increment() {
  35.             count++;  // 使用synchronized确保原子性
  36.         }
  38.         public synchronized int getCount() {
  39.             return count;
  40.         }
  41.     }
  43.     // 测试计数器
  44.     private static void runCounterTest(String name, Object counter) throws InterruptedException {
  45.         // 创建多个线程同时增加计数器
  46.         Thread[] threads = new Thread[10];
  47.         for (int i = 0; i < threads.length; i++) {
  48.             threads[i] = new Thread(() -> {
  49.                 for (int j = 0; j < 10000; j++) {
  50.                     if (counter instanceof UnsafeCounter) {
  51.                         ((UnsafeCounter) counter).increment();
  52.                     } else if (counter instanceof SafeCounter) {
  53.                         ((SafeCounter) counter).increment();
  54.                     }
  55.                 }
  56.             });
  57.             threads[i].start();
  58.         }
  60.         // 等待所有线程完成
  61.         for (Thread thread : threads) {
  62.             thread.join();
  63.         }
  65.         // 检查最终结果
  66.         int finalCount = 0;
  67.         if (counter instanceof UnsafeCounter) {
  68.             finalCount = ((UnsafeCounter) counter).getCount();
  69.         } else if (counter instanceof SafeCounter) {
  70.             finalCount = ((SafeCounter) counter).getCount();
  71.         }
  73.         System.out.println(name + ": 预期结果=100000, 实际结果=" + finalCount + 
  74.                 (finalCount == 100000 ? " (正确)" : " (错误)"));
  75.     }
  76. }

📌 提示： 在上面的例子中，线程不安全的计数器通常会得到小于预期的结果，因为多个线程同时执行count++操作时会发生数据竞争。

1.2 线程安全的重要性

线程安全对于多线程程序至关重要，原因包括：

  1. 数据一致性：确保程序的数据状态始终保持一致，不会出现数据损坏或不一致的情况。

  2. 正确性：确保程序的行为符合预期，产生正确的结果。

  3. 避免难以调试的问题：线程安全问题通常表现为间歇性故障，很难复现和调试。

  4. 提高可靠性：线程安全的程序在各种并发场景下都能稳定运行，提高整体可靠性。

2. 共享资源访问的竞态条件

2.1 什么是竞态条件

竞态条件(Race Condition)是指当多个线程同时访问共享资源，并且尝试同时修改该资源时，最终的结果依赖于线程执行的顺序和时机，导致结果不可预测。

在多线程环境下，竞态条件主要出现在以下场景：

  1. 读-改-写 操作序列不是原子的，可能被其他线程中断
  2. 检查再执行 模式，在检查和执行之间状态可能发生变化
  3. 多步骤操作 不是以原子方式执行的

2.2 竞态条件示例

  1. package org.devlive.tutorial.multithreading.chapter04;
  3. /**
  4.  * 竞态条件示例
  5.  */
  6. public class RaceConditionDemo {
  8.     // 银行账户类（线程不安全）
  9.     static class BankAccount {
  10.         private int balance;  // 余额
  12.         public BankAccount(int initialBalance) {
  13.             this.balance = initialBalance;
  14.         }
  16.         // 存款方法
  17.         public void deposit(int amount) {
  18.             // 读取余额
  19.             int current = balance;
  21.             // 模拟处理延迟，使竞态条件更容易出现
  22.             try {
  23.                 Thread.sleep(1);
  24.             } catch (InterruptedException e) {
  25.                 Thread.currentThread().interrupt();
  26.             }
  28.             // 计算新余额并更新
  29.             balance = current + amount;
  30.         }
  32.         // 取款方法
  33.         public void withdraw(int amount) {
  34.             // 读取余额
  35.             int current = balance;
  37.             // 模拟处理延迟
  38.             try {
  39.                 Thread.sleep(1);
  40.             } catch (InterruptedException e) {
  41.                 Thread.currentThread().interrupt();
  42.             }
  44.             // 只有余额充足才能取款
  45.             if (current >= amount) {
  46.                 // 计算新余额并更新
  47.                 balance = current - amount;
  48.             }
  49.         }
  51.         public int getBalance() {
  52.             return balance;
  53.         }
  54.     }
  56.     public static void main(String[] args) throws InterruptedException {
  57.         // 创建账户，初始余额为1000
  58.         BankAccount account = new BankAccount(1000);
  60.         // 创建多个存款线程，每个存款100
  61.         Thread[] depositThreads = new Thread[5];
  62.         for (int i = 0; i < depositThreads.length; i++) {
  63.             depositThreads[i] = new Thread(() -> {
  64.                 for (int j = 0; j < 10; j++) {
  65.                     account.deposit(100);
  66.                 }
  67.             });
  68.         }
  70.         // 创建多个取款线程，每个取款100
  71.         Thread[] withdrawThreads = new Thread[5];
  72.         for (int i = 0; i < withdrawThreads.length; i++) {
  73.             withdrawThreads[i] = new Thread(() -> {
  74.                 for (int j = 0; j < 10; j++) {
  75.                     account.withdraw(100);
  76.                 }
  77.             });
  78.         }
  80.         // 启动所有线程
  81.         System.out.println("启动线程，模拟账户并发操作...");
  82.         System.out.println("初始余额: " + account.getBalance());
  84.         for (Thread t : depositThreads) t.start();
  85.         for (Thread t : withdrawThreads) t.start();
  87.         // 等待所有线程完成
  88.         for (Thread t : depositThreads) t.join();
  89.         for (Thread t : withdrawThreads) t.join();
  91.         // 检查最终余额
  92.         System.out.println("所有操作完成后余额: " + account.getBalance());
  93.         System.out.println("预期余额: 1000 (初始值) + 5*10*100 (存款) - 5*10*100 (取款) = 1000");
  94.     }
  95. }

在上面的示例中，银行账户的存取款操作不是原子的，由于竞态条件，最终的余额可能不等于预期的1000。

⚠️ 警告： 竞态条件是多线程编程中最常见也是最危险的问题之一，因为它可能导致不可重现的错误和数据损坏。

2.3 竞态条件的类型

读-改-写竞态条件

  1. // 线程A和线程B同时执行这段代码:
  2. int temp = counter;  // 读
  3. temp = temp + 1;     // 改
  4. counter = temp;      // 写

如果counter初始值为0，预期两个线程执行后counter应该是2，但可能会出现以下情况：

  1. 线程A读取counter为0
  2. 线程B读取counter为0
  3. 线程A计算temp + 1 = 1并更新counter为1
  4. 线程B计算temp + 1 = 1并更新counter为1

最终结果是1，而不是预期的2。

检查再执行竞态条件

  1. // 方法示例：转账
  2. public void transfer(Account to, double amount) {
  3.     if (this.balance >= amount) {  // 检查
  4.         this.balance -= amount;    // 执行
  5.         to.balance += amount;      // 执行
  6.     }
  7. }

如果两个线程同时从同一账户转账，且账户余额仅足够一次转账，可能会出现以下情况：

  1. 线程A检查余额足够并准备转账
  2. 线程B检查余额足够并准备转账
  3. 线程A执行转账，减少余额
  4. 线程B执行转账，再次减少余额，可能导致余额为负

3. 线程安全问题的表现形式

3.1 原子性问题

当一个操作本应该是不可分割的，但在多线程环境下被拆分执行时，就会出现原子性问题。Java中许多看起来是一条语句的操作实际上并不是原子的。

  1. package org.devlive.tutorial.multithreading.chapter04;
  3. import java.util.concurrent.atomic.AtomicInteger;
  5. /**
  6.  * 原子性问题示例
  7.  */
  8. public class AtomicityProblemDemo {
  9.     public static void main(String[] args) throws InterruptedException {
  10.         // 测试非原子操作
  11.         testCounter("非原子计数器", new NonAtomicCounter());
  13.         // 测试使用原子变量的计数器
  14.         testCounter("原子计数器", new AtomicCounter());
  15.     }
  17.     // 非原子计数器
  18.     static class NonAtomicCounter {
  19.         private int count = 0;
  21.         public void increment() {
  22.             count++; // 看似简单，实际上是读-改-写三步操作
  23.         }
  25.         public int getCount() {
  26.             return count;
  27.         }
  28.     }
  30.     // 使用原子变量的计数器
  31.     static class AtomicCounter {
  32.         private AtomicInteger count = new AtomicInteger(0);
  34.         public void increment() {
  35.             count.incrementAndGet(); // 原子操作
  36.         }
  38.         public int getCount() {
  39.             return count.get();
  40.         }
  41.     }
  43.     // 测试计数器
  44.     private static void testCounter(String name, Object counter) throws InterruptedException {
  45.         final int NUM_THREADS = 10;
  46.         final int ITERATIONS = 100000;
  48.         // 创建并启动多个线程递增计数器
  49.         Thread[] threads = new Thread[NUM_THREADS];
  50.         for (int i = 0; i < NUM_THREADS; i++) {
  51.             threads[i] = new Thread(() -> {
  52.                 for (int j = 0; j < ITERATIONS; j++) {
  53.                     if (counter instanceof NonAtomicCounter) {
  54.                         ((NonAtomicCounter) counter).increment();
  55.                     } else if (counter instanceof AtomicCounter) {
  56.                         ((AtomicCounter) counter).increment();
  57.                     }
  58.                 }
  59.             });
  60.             threads[i].start();
  61.         }
  63.         // 等待所有线程完成
  64.         for (Thread thread : threads) {
  65.             thread.join();
  66.         }
  68.         // 检查结果
  69.         int finalCount = 0;
  70.         if (counter instanceof NonAtomicCounter) {
  71.             finalCount = ((NonAtomicCounter) counter).getCount();
  72.         } else if (counter instanceof AtomicCounter) {
  73.             finalCount = ((AtomicCounter) counter).getCount();
  74.         }
  76.         int expectedCount = NUM_THREADS * ITERATIONS;
  77.         System.out.println(name + " - 预期结果: " + expectedCount + ", 实际结果: " + finalCount +
  78.                 (finalCount == expectedCount ? " (正确)" : " (错误)"));
  79.     }
  80. }

📌 提示： 在Java中，以下操作都不是原子的：

  • 递增/递减操作（如i++, i--）
  • 复合赋值操作（如i += 5）
  • 在volatile变量上执行的非赋值操作

3.2 可见性问题

可见性问题指的是一个线程对共享变量的修改，另一个线程可能看不到这个修改的最新值。这通常是由于CPU缓存、编译器优化或指令重排序导致的。

  1. package org.devlive.tutorial.multithreading.chapter04;
  3. /**
  4.  * 可见性问题示例
  5.  */
  6. public class VisibilityProblemDemo {
  7.     // 没有volatile修饰的标志变量
  8.     private static boolean stopRequested = false;
  10.     // 使用volatile修饰的标志变量
  11.     private static volatile boolean volatileStopRequested = false;
  13.     public static void main(String[] args) throws InterruptedException {
  14.         // 测试没有volatile的情况
  15.         testVisibility(false);
  17.         // 测试使用volatile的情况
  18.         testVisibility(true);
  19.     }
  21.     private static void testVisibility(boolean useVolatile) throws InterruptedException {
  22.         System.out.println("\n======== " + (useVolatile ? "使用volatile" : "不使用volatile") + " ========");
  24.         // 重置标志
  25.         stopRequested = false;
  26.         volatileStopRequested = false;
  28.         // 创建工作线程，不断检查标志变量
  29.         Thread workerThread = new Thread(() -> {
  30.             long i = 0;
  31.             System.out.println("工作线程开始执行...");
  33.             // 根据参数选择使用哪个标志变量
  34.             if (useVolatile) {
  35.                 while (!volatileStopRequested) {
  36.                     i++;
  37.                 }
  38.             } else {
  39.                 while (!stopRequested) {
  40.                     i++;
  41.                 }
  42.             }
  44.             System.out.println("工作线程检测到停止信号，循环次数：" + i);
  45.         });
  47.         workerThread.start();
  49.         // 主线程等待一会儿
  50.         Thread.sleep(1000);
  52.         System.out.println("主线程设置停止信号...");
  54.         // 设置停止标志
  55.         if (useVolatile) {
  56.             volatileStopRequested = true;
  57.         } else {
  58.             stopRequested = true;
  59.         }
  61.         // 等待工作线程结束
  62.         workerThread.join(5000);
  64.         // 检查线程是否还活着
  65.         if (workerThread.isAlive()) {
  66.             System.out.println("工作线程仍在运行！可能存在可见性问题");
  67.             workerThread.interrupt(); // 强制中断
  68.         } else {
  69.             System.out.println("工作线程正确终止");
  70.         }
  71.     }
  72. }

在不使用volatile的情况下，由于可见性问题，工作线程可能无法看到主线程对stopRequested的修改，导致它无限循环下去。

📌 提示： Java提供了volatile关键字来解决可见性问题。被volatile修饰的变量，对它的读写都会直接在主内存中进行，而不是在CPU缓存中。

3.3 有序性问题

有序性问题是指程序的执行顺序可能与代码编写的顺序不同。这是由于编译器优化、CPU指令重排序等导致的。在单线程环境下，重排序后的执行结果与顺序执行的结果相同，但在多线程环境下可能导致问题。

  1. package org.devlive.tutorial.multithreading.chapter04;
  3. /**
  4.  * 有序性问题示例
  5.  */
  6. public class OrderingProblemDemo {
  7.     private static int x = 0, y = 0;
  8.     private static int a = 0, b = 0;
  10.     public static void main(String[] args) throws InterruptedException {
  11.         int iterations = 0;
  12.         int abnormalResults = 0;
  14.         // 多次运行测试，统计出现重排序的次数
  15.         final int TEST_COUNT = 100000;
  16.         System.out.println("开始测试重排序问题，运行 " + TEST_COUNT + " 次...");
  18.         for (int i = 0; i < TEST_COUNT; i++) {
  19.             iterations++;
  21.             // 重置变量
  22.             x = 0; y = 0;
  23.             a = 0; b = 0;
  25.             // 创建线程1
  26.             Thread thread1 = new Thread(() -> {
  27.                 a = 1;  // 语句1
  28.                 x = b;  // 语句2
  29.             });
  31.             // 创建线程2
  32.             Thread thread2 = new Thread(() -> {
  33.                 b = 1;  // 语句3
  34.                 y = a;  // 语句4
  35.             });
  37.             // 启动线程
  38.             thread1.start();
  39.             thread2.start();
  41.             // 等待线程结束
  42.             thread1.join();
  43.             thread2.join();
  45.             // 检查结果
  46.             if (x == 0 && y == 0) {
  47.                 abnormalResults++;
  48.                 // 因为出现频率较低，只在前几次出现时打印详细信息
  49.                 if (abnormalResults <= 10) {
  50.                     System.out.println("检测到可能的指令重排序: x=" + x + ", y=" + y);
  51.                 }
  52.             }
  53.         }
  55.         // 统计结果
  56.         System.out.println("\n测试完成:");
  57.         System.out.println("总测试次数: " + iterations);
  58.         System.out.println("检测到的异常结果次数(x=0, y=0): " + abnormalResults);
  59.         System.out.println("异常结果比例: " + String.format("%.5f%%", (double)abnormalResults / iterations * 100));
  61.         System.out.println("\n分析:");
  62.         if (abnormalResults > 0) {
  63.             System.out.println("检测到可能的指令重排序。在某些情况下，两个线程中的操作可能被重排序，导致x=0且y=0。");
  64.             System.out.println("这表明Java内存模型允许某些指令重排序，可能影响多线程程序的执行结果。");
  65.         } else {
  66.             System.out.println("未检测到明显的指令重排序。这可能是由于:");
  67.             System.out.println("1. 当前硬件和JVM实现中不太容易观察到这种重排序");
  68.             System.out.println("2. 测试次数不够多");
  69.             System.out.println("但这并不意味着指令重排序不存在。在复杂的多线程程序中，它仍然可能导致问题。");
  70.         }
  71.     }
  72. }

在上面的例子中，如果没有指令重排序，理论上x和y不可能同时为0。但由于指令重排序的存在，它们确实可能同时为0。

📌 提示： Java内存模型通过happens-before关系保证有序性。此外，volatile、synchronized和final关键字也能在一定程度上阻止有害的指令重排序。

4. 实战案例：商品库存管理

下面我们通过一个商品库存管理的实战案例，来综合应用所学的线程安全知识：

  1. package org.devlive.tutorial.multithreading.chapter04;
  3. import java.util.HashMap;
  4. import java.util.Map;
  5. import java.util.concurrent.ConcurrentHashMap;
  6. import java.util.concurrent.atomic.AtomicInteger;
  7. import java.util.concurrent.locks.ReadWriteLock;
  8. import java.util.concurrent.locks.ReentrantReadWriteLock;
  10. /**
  11.  * 商品库存管理实战案例
  12.  */
  13. public class InventoryManagementDemo {
  15.     // 线程不安全的库存管理
  16.     static class UnsafeInventory {
  17.         private final Map<String, Integer> productStock = new HashMap<>();
  19.         // 添加或更新库存
  20.         public void updateStock(String productId, int quantity) {
  21.             Integer currentQuantity = productStock.get(productId);
  22.             if (currentQuantity == null) {
  23.                 productStock.put(productId, quantity);
  24.             } else {
  25.                 productStock.put(productId, currentQuantity + quantity);
  26.             }
  27.         }
  29.         // 减少库存（如果库存不足，返回false）
  30.         public boolean decreaseStock(String productId, int quantity) {
  31.             Integer currentQuantity = productStock.get(productId);
  32.             if (currentQuantity == null || currentQuantity < quantity) {
  33.                 return false;
  34.             }
  36.             productStock.put(productId, currentQuantity - quantity);
  37.             return true;
  38.         }
  40.         // 获取当前库存
  41.         public Integer getStock(String productId) {
  42.             return productStock.getOrDefault(productId, 0);
  43.         }
  45.         // 获取所有商品库存
  46.         public Map<String, Integer> getAllStock() {
  47.             return new HashMap<>(productStock);
  48.         }
  49.     }
  51.     // 使用synchronized的线程安全库存管理
  52.     static class SynchronizedInventory {
  53.         private final Map<String, Integer> productStock = new HashMap<>();
  55.         // 添加或更新库存
  56.         public synchronized void updateStock(String productId, int quantity) {
  57.             Integer currentQuantity = productStock.get(productId);
  58.             if (currentQuantity == null) {
  59.                 productStock.put(productId, quantity);
  60.             } else {
  61.                 productStock.put(productId, currentQuantity + quantity);
  62.             }
  63.         }
  65.         // 减少库存（如果库存不足，返回false）
  66.         public synchronized boolean decreaseStock(String productId, int quantity) {
  67.             Integer currentQuantity = productStock.get(productId);
  68.             if (currentQuantity == null || currentQuantity < quantity) {
  69.                 return false;
  70.             }
  72.             productStock.put(productId, currentQuantity - quantity);
  73.             return true;
  74.         }
  76.         // 获取当前库存
  77.         public synchronized Integer getStock(String productId) {
  78.             return productStock.getOrDefault(productId, 0);
  79.         }
  81.         // 获取所有商品库存
  82.         public synchronized Map<String, Integer> getAllStock() {
  83.             return new HashMap<>(productStock);
  84.         }
  85.     }
  87.     // 使用ConcurrentHashMap和AtomicInteger的线程安全库存管理
  88.     static class ConcurrentInventory {
  89.         private final Map<String, AtomicInteger> productStock = new ConcurrentHashMap<>();
  91.         // 添加或更新库存
  92.         public void updateStock(String productId, int quantity) {
  93.             productStock.computeIfAbsent(productId, k -> new AtomicInteger(0))
  94.                     .addAndGet(quantity);
  95.         }
  97.         // 减少库存（如果库存不足，返回false）
  98.         public boolean decreaseStock(String productId, int quantity) {
  99.             AtomicInteger stock = productStock.get(productId);
  100.             if (stock == null) {
  101.                 return false;
  102.             }
  104.             int currentValue;
  105.             do {
  106.                 currentValue = stock.get();
  107.                 if (currentValue < quantity) {
  108.                     return false;
  109.                 }
  110.             } while (!stock.compareAndSet(currentValue, currentValue - quantity));
  112.             return true;
  113.         }
  115.         // 获取当前库存
  116.         public Integer getStock(String productId) {
  117.             AtomicInteger stock = productStock.get(productId);
  118.             return stock != null ? stock.get() : 0;
  119.         }
  121.         // 获取所有商品库存
  122.         public Map<String, Integer> getAllStock() {
  123.             Map<String, Integer> result = new HashMap<>();
  124.             productStock.forEach((key, value) -> result.put(key, value.get()));
  125.             return result;
  126.         }
  127.     }
  129.     // 使用读写锁的线程安全库存管理
  130.     static class ReadWriteLockInventory {
  131.         private final Map<String, Integer> productStock = new HashMap<>();
  132.         private final ReadWriteLock lock = new ReentrantReadWriteLock();
  134.         // 添加或更新库存（写操作）
  135.         public void updateStock(String productId, int quantity) {
  136.             lock.writeLock().lock();
  137.             try {
  138.                 Integer currentQuantity = productStock.get(productId);
  139.                 if (currentQuantity == null) {
  140.                     productStock.put(productId, quantity);
  141.                 } else {
  142.                     productStock.put(productId, currentQuantity + quantity);
  143.                 }
  144.             } finally {
  145.                 lock.writeLock().unlock();
  146.             }
  147.         }
  149.         // 减少库存（写操作）
  150.         public boolean decreaseStock(String productId, int quantity) {
  151.             lock.writeLock().lock();
  152.             try {
  153.                 Integer currentQuantity = productStock.get(productId);
  154.                 if (currentQuantity == null || currentQuantity < quantity) {
  155.                     return false;
  156.                 }
  158.                 productStock.put(productId, currentQuantity - quantity);
  159.                 return true;
  160.             } finally {
  161.                 lock.writeLock().unlock();
  162.             }
  163.         }
  165.         // 获取当前库存（读操作）
  166.         public Integer getStock(String productId) {
  167.             lock.readLock().lock();
  168.             try {
  169.                 return productStock.getOrDefault(productId, 0);
  170.             } finally {
  171.                 lock.readLock().unlock();
  172.             }
  173.         }
  175.         // 获取所有商品库存（读操作）
  176.         public Map<String, Integer> getAllStock() {
  177.             lock.readLock().lock();
  178.             try {
  179.                 return new HashMap<>(productStock);
  180.             } finally {
  181.                 lock.readLock().unlock();
  182.             }
  183.         }
  184.     }
  186.     public static void main(String[] args) throws InterruptedException {
  187.         // 创建不同类型的库存管理器
  188.         UnsafeInventory unsafeInventory = new UnsafeInventory();
  189.         SynchronizedInventory syncInventory = new SynchronizedInventory();
  190.         ConcurrentInventory concurrentInventory = new ConcurrentInventory();
  191.         ReadWriteLockInventory rwlInventory = new ReadWriteLockInventory();
  193.         // 初始化库存
  194.         String[] products = {"iPhone", "MacBook", "iPad", "AirPods"};
  195.         for (String product : products) {
  196.             unsafeInventory.updateStock(product, 1000);
  197.             syncInventory.updateStock(product, 1000);
  198.             concurrentInventory.updateStock(product, 1000);
  199.             rwlInventory.updateStock(product, 1000);
  200.         }
  202.         // 测试线程不安全的库存管理
  203.         System.out.println("测试线程不安全的库存管理");
  204.         testInventory(unsafeInventory, products);
  206.         // 测试synchronized的库存管理
  207.         System.out.println("\n测试synchronized的库存管理");
  208.         testInventory(syncInventory, products);
  210.         // 测试ConcurrentHashMap的库存管理
  211.         System.out.println("\n测试ConcurrentHashMap的库存管理");
  212.         testInventory(concurrentInventory, products);
  214.         // 测试ReadWriteLock的库存管理
  215.         System.out.println("\n测试ReadWriteLock的库存管理");
  216.         testInventory(rwlInventory, products);
  217.     }
  219.     private static void testInventory(Object inventory, String[] products) throws InterruptedException {
  220.         // 创建多个购买线程（减少库存）
  221.         Thread[] buyThreads = new Thread[10];
  222.         for (int i = 0; i < buyThreads.length; i++) {
  223.             buyThreads[i] = new Thread(() -> {
  224.                 for (int j = 0; j < 100; j++) {
  225.                     String product = products[j % products.length];
  226.                     boolean success = false;
  228.                     if (inventory instanceof UnsafeInventory) {
  229.                         success = ((UnsafeInventory) inventory).decreaseStock(product, 1);
  230.                     } else if (inventory instanceof SynchronizedInventory) {
  231.                         success = ((SynchronizedInventory) inventory).decreaseStock(product, 1);
  232.                     } else if (inventory instanceof ConcurrentInventory) {
  233.                         success = ((ConcurrentInventory) inventory).decreaseStock(product, 1);
  234.                     } else if (inventory instanceof ReadWriteLockInventory) {
  235.                         success = ((ReadWriteLockInventory) inventory).decreaseStock(product, 1);
  236.                     }
  238.                     if (!success) {
  239.                         System.out.println("购买失败: " + product + " - 库存不足");
  240.                     }
  241.                 }
  242.             });
  243.         }
  245.         // 创建多个补货线程（增加库存）
  246.         Thread[] restockThreads = new Thread[5];
  247.         for (int i = 0; i < restockThreads.length; i++) {
  248.             restockThreads[i] = new Thread(() -> {
  249.                 for (int j = 0; j < 40; j++) {
  250.                     String product = products[j % products.length];
  252.                     if (inventory instanceof UnsafeInventory) {
  253.                         ((UnsafeInventory) inventory).updateStock(product, 5);
  254.                     } else if (inventory instanceof SynchronizedInventory) {
  255.                         ((SynchronizedInventory) inventory).updateStock(product, 5);
  256.                     } else if (inventory instanceof ConcurrentInventory) {
  257.                         ((ConcurrentInventory) inventory).updateStock(product, 5);
  258.                     } else if (inventory instanceof ReadWriteLockInventory) {
  259.                         ((ReadWriteLockInventory) inventory).updateStock(product, 5);
  260.                     }
  261.                 }
  262.             });
  263.         }
  265.         // 创建多个查询线程（读取库存）
  266.         Thread[] queryThreads = new Thread[20];
  267.         for (int i = 0; i < queryThreads.length; i++) {
  268.             queryThreads[i] = new Thread(() -> {
  269.                 for (int j = 0; j < 50; j++) {
  270.                     String product = products[j % products.length];
  271.                     int stock = 0;
  273.                     if (inventory instanceof UnsafeInventory) {
  274.                         stock = ((UnsafeInventory) inventory).getStock(product);
  275.                     } else if (inventory instanceof SynchronizedInventory) {
  276.                         stock = ((SynchronizedInventory) inventory).getStock(product);
  277.                     } else if (inventory instanceof ConcurrentInventory) {
  278.                         stock = ((ConcurrentInventory) inventory).getStock(product);
  279.                     } else if (inventory instanceof ReadWriteLockInventory) {
  280.                         stock = ((ReadWriteLockInventory) inventory).getStock(product);
  281.                     }
  283.                     // 不打印库存信息，避免输出过多
  284.                 }
  285.             });
  286.         }
  288.         // 记录开始时间
  289.         long startTime = System.currentTimeMillis();
  291.         // 启动所有线程
  292.         for (Thread t : buyThreads) t.start();
  293.         for (Thread t : restockThreads) t.start();
  294.         for (Thread t : queryThreads) t.start();
  296.         // 等待所有线程完成
  297.         for (Thread t : buyThreads) t.join();
  298.         for (Thread t : restockThreads) t.join();
  299.         for (Thread t : queryThreads) t.join();
  301.         // 计算耗时
  302.         long endTime = System.currentTimeMillis();
  304.         // 输出最终库存和执行时间
  305.         System.out.println("执行时间: " + (endTime - startTime) + "ms");
  306.         System.out.println("最终库存:");
  308.         Map<String, Integer> finalStock = null;
  309.         if (inventory instanceof UnsafeInventory) {
  310.             finalStock = ((UnsafeInventory) inventory).getAllStock();
  311.         } else if (inventory instanceof SynchronizedInventory) {
  312.             finalStock = ((SynchronizedInventory) inventory).getAllStock();
  313.         } else if (inventory instanceof ConcurrentInventory) {
  314.             finalStock = ((ConcurrentInventory) inventory).getAllStock();
  315.         } else if (inventory instanceof ReadWriteLockInventory) {
  316.             finalStock = ((ReadWriteLockInventory) inventory).getAllStock();
  317.         }
  319.         if (finalStock != null) {
  320.             for (String product : products) {
  321.                 System.out.println(product + ": " + finalStock.get(product));
  322.             }
  323.         }
  325.         // 验证库存一致性
  326.         int expectedBaseline = 1000; // 初始库存
  327.         int buyOps = buyThreads.length * 100; // 总购买操作
  328.         int restockOps = restockThreads.length * 40 * 5; // 总补货操作
  330.         // 平均到每种商品
  331.         int opsPerProduct = (buyOps - restockOps) / products.length;
  332.         int expectedStock = expectedBaseline - opsPerProduct;
  334.         System.out.println("\n库存检查:");
  335.         System.out.println("每种商品预期变化: " + opsPerProduct + " (负数表示库存减少)");
  336.         System.out.println("预期最终库存: " + expectedStock);
  337.     }
  338. }

在这个案例中，我们实现了四种不同的库存管理方式，并比较了它们在多线程环境下的表现：

  1. 不安全的库存管理（使用HashMap）
  2. 使用synchronized的库存管理
  3. 使用ConcurrentHashMap和AtomicInteger的库存管理
  4. 使用读写锁的库存管理

测试结果可能显示线程不安全的实现存在数据不一致的问题，而其他三种线程安全的实现能够保证数据的正确性，但性能各有差异。

常见问题与解决方案

问题1：多个操作需要作为一个原子单元执行

问题描述： 有时我们需要确保多个操作作为一个不可分割的单元执行，以保持数据的一致性。

解决方案： 使用锁（synchronized或显式锁）来确保互斥访问。

  1. // 使用synchronized
  2. public synchronized boolean transferMoney(Account from, Account to, double amount) {
  3.     if (from.getBalance() < amount) {
  4.         return false;
  5.     }
  6.     from.debit(amount);
  7.     to.credit(amount);
  8.     return true;
  9. }
  11. // 使用显式锁
  12. public boolean transferMoney(Account from, Account to, double amount) {
  13.     lock.lock();
  14.     try {
  15.         if (from.getBalance() < amount) {
  16.             return false;
  17.         }
  18.         from.debit(amount);
  19.         to.credit(amount);
  20.         return true;
  21.     } finally {
  22.         lock.unlock();
  23.     }
  24. }

问题2：频繁读取但很少修改的数据

问题描述： 有些数据结构被频繁读取但很少修改，使用互斥锁会导致读操作也被阻塞，影响性能。

解决方案： 使用读写锁（ReadWriteLock）允许多个读操作并发执行，而写操作依然互斥。

  1. private final Map<String, Product> productCatalog = new HashMap<>();
  2. private final ReadWriteLock lock = new ReentrantReadWriteLock();
  4. // 读操作（多个线程可以同时读取）
  5. public Product getProduct(String productId) {
  6.     lock.readLock().lock();
  7.     try {
  8.         return productCatalog.get(productId);
  9.     } finally {
  10.         lock.readLock().unlock();
  11.     }
  12. }
  14. // 写操作（互斥访问）
  15. public void updateProduct(Product product) {
  16.     lock.writeLock().lock();
  17.     try {
  18.         productCatalog.put(product.getId(), product);
  19.     } finally {
  20.         lock.writeLock().unlock();
  21.     }
  22. }

问题3：在检查和执行之间保持一致性

问题描述： 在执行某个操作前需要先检查条件，但在多线程环境下，检查和执行之间的状态可能已经变化。

解决方案： 使用锁确保检查和执行的原子性，或者使用原子变量的CAS操作。

  1. // 使用锁确保原子性
  2. public synchronized boolean decrementIfPositive(int[] array, int index) {
  3.     if (array[index] > 0) {
  4.         array[index]--;
  5.         return true;
  6.     }
  7.     return false;
  8. }
  10. // 使用AtomicIntegerArray的CAS操作
  11. private final AtomicIntegerArray atomicArray = new AtomicIntegerArray(size);
  13. public boolean decrementIfPositive(int index) {
  14.     while (true) {
  15.         int current = atomicArray.get(index);
  16.         if (current <= 0) {
  17.             return false;
  18.         }
  19.         if (atomicArray.compareAndSet(index, current, current - 1)) {
  20.             return true;
  21.         }
  22.         // 如果CAS失败，循环重试
  23.     }
  24. }

小结

在本章中，我们深入探讨了线程安全问题，学习了以下关键内容：

  1. 线程安全的概念：线程安全指在多线程环境下，程序能够正确处理共享资源，确保数据的一致性和正确性。

  2. 竞态条件：当多个线程同时访问共享资源并尝试修改时，结果依赖于线程执行的顺序和时机，可能导致不可预测的结果。

  3. 线程安全问题的三种表现：

    • 原子性问题：操作被拆分执行，导致数据不一致
    • 可见性问题：一个线程的修改对其他线程不可见
    • 有序性问题：指令重排序导致的执行顺序问题

  4. 解决方案：

    • 使用synchronized关键字确保互斥访问
    • 使用原子变量（如AtomicInteger）进行原子操作
    • 使用显式锁（如ReentrantLock）提供更灵活的锁机制
    • 使用读写锁（ReadWriteLock）优化读多写少的场景
    • 使用线程安全的集合类（如ConcurrentHashMap）

  5. 实战应用：通过商品库存管理案例，我们实践了不同的线程安全实现方式，并比较了它们的性能和正确性。

理解和解决线程安全问题是多线程编程的核心挑战。通过合理使用Java提供的同步工具，我们可以开发出高效、可靠的多线程应用程序。

在下一章，我们将详细探讨synchronized关键字的使用和底层实现原理。

本章节源代码地址为 https://github.com/qianmoQ/tutorial/tree/main/java-multithreading-tutorial/src/main/java/org/devlive/tutorial/multithreading/chapter04