并发编程——线程安全问题

前言

众所周知，Java中提供了许多线程安全的集合、操作类等，可能我们直接拿来用就行了，但是没搞懂什么时候用，为什么要用？那么这篇文章会帮助你！

线程不安全情况

线程不安全（Thread-unsafe）指的是在多线程环境中，某个操作或某些操作序列在多个线程并发执行时，不能保证数据的正确性和一致性，可能会产生意料之外的结果。简单来说，如果一个对象或资源在多个线程同时访问时，没有进行适当的同步措施，就可能导致线程不安全。最典型的案例：商品SKU超卖问题。

以下是线程不安全的一些常见原因和表现：

竞态条件（Race Conditions）：当多个线程试图同时访问和修改同一数据时，如果没有适当的同步机制，那么最后的结果可能取决于线程的执行顺序，这种情况称为竞态条件。
数据不一致（Data Inconsistency）：由于多个线程可以同时修改同一数据，如果没有适当的同步，可能会导致数据处于不一致的状态。
内存可见性（Memory Visibility）：一个线程对共享变量的修改可能对其他线程不可见，这可能导致其他线程读取到旧值。

举个计数器的例子，如下

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class JUCDemo {
    private final static int NUM_THREADS = 5;
    private static int count = 0;

    public static void main(String[] args) {
        ExecutorService executorService = Executors.newFixedThreadPool(NUM_THREADS);
        for (int i = 0; i < NUM_THREADS; i++) {
            executorService.execute(() -> {
                for (int j = 0; j < 1000; j++) {
                    count++;
                }
            });
        }
        //关闭线程池
        executorService.shutdown();
        //判断线程池是否终止，如果终止则打印count
        while (!executorService.isTerminated()) {
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        System.out.println("count = " + count);
    }
}

通过Executor创建一个固定大小的线程池，对线程池循环执行5次，并将每个线程执行方法中count共享变量加1000次，大家猜想一下，最终打印的这个count会是多少呢？

答案是：小于等于5000

如果能想到小于等于5000，证明你的并发思想还是有点功底的，那么现在来解释一下，为什么会出现小于5000的情况。

这里不得不引申一下Java的内存模型了：

对于count变量来说，它是一个共享变量，储存在主内存中，当线程执行，每个线程会从主内存中copy一份副本到自己的本地内存中，所以每个线程中修改的不是主内存里的count，而是自己本地内存（也叫工作内存）的副本，修改完后，才会通过IO进行同步主内存里的count值。

如果线程1和线程2同时从主内存中copy一份count的副本，这时线程1和线程2本地内存中都是0，在各自的本地内存中执行加1操作后，同步到主内存，都是1，这就出现漏加的情况了，本来两次执行操作可以加2，结果现在只加了1。

如何防止线程不安全

这里就要说一下Java提供的那些线程安全的集合、操作类了，因为它们底层都做了加锁处理。最常见的就是synchronized 关键字和ReentrantLock类。

（1）synchronized ：

synchronized 关键字: Java 中的每个对象都可以作为锁，使用它可以同步方法或代码块。当一个线程访问一个对象的 synchronized 方法或代码块时，它会获得该对象的内置锁，其他线程将无法同时访问该对象的任何 synchronized 方法或代码块，阻塞其他线程，当前线程执行完毕后才会释放锁，其他线程才会有机会获取锁。通俗的来讲：就好比共享资源是个公共厕所，好多人在厕所前排队，一个人进去后进行上锁，其他人只能等待。

下面是改造后的代码，这样就能保证count不会出现漏加的情况

Object lock = new Object(); // 创建一个锁对象
public static void main(String[] args) {
  ExecutorService executorService = Executors.newFixedThreadPool(NUM_THREADS);
  for (int i = 0; i < NUM_THREADS; i++) {
      executorService.execute(() -> {
          for (int j = 0; j < 1000; j++) {
              synchronized (lock) {
                  count++;
              }
          }
      });
  }
}

synchronized 锁在JDK1.6之前，它的性能是很差的，但是经过JDK1.6之后，经过优化，它的性能得到很大提升。特别是加入了从偏向锁——轻量级锁——重量级锁的加锁过程，基本上synchronized 和ReentrantLock的性能是差不多的。

注意：synchronized 只能用在单体项目中，synchronized只能保证单个jvm内部的多个线程之间的互斥，如果在集群部署模型下，是会失效的，就要采用分布式锁来保证，比如：Redis的SETNX命令，zookeeper分布式唯一id等，后续会出一期关于使用Redis实现分布式锁的文章。

（2）ReentrantLock：

ReentrantLock其实就是基于AQS实现的一个可重入锁，支持公平和非公平两种方式。内部实现依靠一个state 变量和两个等待队列:同步队列和等待队列。利用 CAS自旋锁修改state来争抢锁，争抢不到则入同步队列等待，同步队列是一个双向链表。条件不满足时候则入等待队列等待，是个单向链表。

ReentrantLock因为是通过CAS自旋来实现的，不会阻塞线程，而是一直循环尝试获取锁，所以消耗较多CPU资源。

这里不再过多延伸，感兴趣的小伙伴可以查一下相关资料。下面是改造后的代码。

private static Lock lock = new ReentrantLock();// 创建一个ReentrantLock实例
public static void main(String[] args) {
  ExecutorService executorService = Executors.newFixedThreadPool(NUM_THREADS);
  for (int i = 0; i < NUM_THREADS; i++) {
    executorService.execute(() -> {
        for (int j = 0; j < 1000; j++) {
            //加锁
            lock.lock();
            try {
                count++;
            } finally {
                //锁释放
                lock.unlock();
            }
        }
    });
  }
}

（3）原子操作：

Java为我们提供了一些原子操作类，在java.util.concurrent.atomic包下，来保证线程安全。常用的有：

//基本类型
AtomicBoolean  //原子地更新布尔值。
AtomicInteger  //原子地更新整数值。
AtomicLong     //原子地更新长整数值。
//数组原子类
AtomicIntegerArray    //原子地更新整型数组里的元素。
AtomicLongArray       //原子地更新长整型数组里的元素。
AtomicReferenceArray  //原子地更新引用类型数组里的元素。

为了解决上面案例的问题，下面就用AtomicInteger来演示

//创建原子操作类，默认为0
private static AtomicInteger count = new AtomicInteger(0);
publicstatic void main(String[] args) {
      // 创建一个固定大小的线程池
      ExecutorService executor = Executors.newFixedThreadPool(NUM_THREADS);

      // 创建任务并提交给线程池执行
      for (int i = 0; i < NUM_THREADS; i++) {
          executor.execute(() -> {
              for (int j = 0; j < INCREMENT_TIMES; j++) {
                  count.incrementAndGet(); // 原子性增加count的值
              }
          });
      }
      //关闭线程池
      executorService.shutdown();
      //判断线程池是否终止，如果终止则打印count
      while (!executorService.isTerminated()) {
          try {
              Thread.sleep(100);
          } catch (InterruptedException e) {
              e.printStackTrace();
          }
      }
      System.out.println("count = " + count);
  }