Core Java 并发：理解并发概念

内容简介：Core Java 并发：理解并发概念

（给ImportNew加星标，提高 Java 技能）

来自：唐尤华

https://dzone.com/refcardz/core-java-concurrency

1. 简介

从诞生开始，Java 就支持线程、锁等关键的并发概念。这篇文章旨在为使用了多线程的 Java 开发者理解 Core Java 中的并发概念以及使用方法。

2. 概念

表1 并发概念

2.1 竞争条件

多个线程对共享资源执行一系列操作，根据每个线程的操作顺序可能存在几种结果，这时出现竞争条件。下面的代码不是线程安全的，而且可以不止一次地初始化 value，因为 check-then-act（检查 null，然后初始化），所以延迟初始化的字段不具备原子性：

class Lazy <T> {
  private volatile T value;
  T get() {
    if (value == null)
      value = initialize();
    return value;
  }
}

2.2 数据竞争

两个或多个线程试图访问同一个非 final 变量并且不加上同步机制，这时会发生数据竞争。没有同步机制可能导致这样的情况，线程执行过程中做出其他线程无法看到的更改，因而导致读到修改前的数据。这样反过来可能又会导致无限循环、破坏数据结构或得到错误的计算结果。下面这段代码可能会无限循环，因为读线程可能永远不知道写线程所做的更改：

class Waiter implements Runnable {
  private boolean shouldFinish;
  void finish() { shouldFinish = true; }
  public void run() {
    long iteration = 0;
    while (!shouldFinish) {
      iteration++;
    }
    System.out.println("Finished after: " + iteration);
  }
}

class DataRace {
  public static void main(String[] args) throws InterruptedException {
    Waiter waiter = new Waiter();
    Thread waiterThread = new Thread(waiter);
    waiterThread.start();
    waiter.finish();
    waiterThread.join();
  }
}

3. Java 内存模型：happens-before 关系

Java 内存模型定义基于一些操作，比如读写字段、 Monitor 同步等。这些操作可以按照 happens-before 关系进行排序。这种关系可用来推断一个线程何时看到另一个线程的操作结果，以及构成一个程序同步后的所有信息。

happens-before 关系具备以下特性：

• 在线程开始所有操作前调用 Thread#start

• 在获取 Monitor 前，释放该 Monitor

• 在读取 volatile 变量前，对该变量执行一次写操作

• 在写入 final 变量前，确保在对象引用已存在

• 线程中的所有操作应在 Thread#join 返回之前完成

4. 标准同步特性

4.1 synchronized 关键字

使用 synchronized 关键字可以防止不同线程同时执行相同代码块。由于进入同步执行的代码块之前加锁，受该锁保护的数据可以在排他模式下操作，从而让操作具备原子性。此外，其他线程在获得相同的锁后也能看到操作结果。

class AtomicOperation {
  private int counter0;
  private int counter1;
  void increment() {
    synchronized (this) {
      counter0++;
      counter1++;
    }
  }
}

也可以在方法上加 synchronized 关键字。

表2 当整个方法都标记 synchronized 时使用的 Monitor

锁是可重入的。如果线程已经持有锁，它可以再次成功地获得该锁。

class Reentrantcy {
  synchronized void doAll() {
    doFirst();
    doSecond();
  }
  synchronized void doFirst() {
    System.out.println("First operation is successful.");
  }
  synchronized void doSecond() {
    System.out.println("Second operation is successful.");
  }
}

竞争的程度对获取 Monitor 的方式有影响：

表3： Monitor 状态

4.2 wait/notify

wait/notify/notifyAll 方法在 Object 类中声明。如果之前设置了超时，线程进入 WAITING 或 TIMED_WAITING 状态前保持 wait状态。要唤醒一个线程，可以执行下列任何操作：

• 另一个线程调用 notify 将唤醒任意一个在 Monitor 上等待的线程。

• 另一个线程调用 notifyAll 将唤醒所有在等待 Monitor 上等待的线程。

• 调用 Thread#interrupt 后会抛出 InterruptedException 异常。

最常见的模式是条件循环：

class ConditionLoop {
  private boolean condition;
  synchronized void waitForCondition() throws InterruptedException {
    while (!condition) {
      wait();
    }
  }
  synchronized void satisfyCondition() {
    condition = true;
    notifyAll();
  }
}

• 请记住，在对象上调用 wait/notify/notifyAll，需要首先获得该对象的锁

• 在检查等待条件的循环中保持等待：这解决了另一个线程在等待开始之前即满足条件时的计时问题。 此外，这样做还可以让你的代码免受可能（也的确会）发生的虚假唤醒

• 在调用 notify/notifyAll 前，要确保满足等待条件。如果不这样做会引发通知，然而没有线程能够避免等待循环

4.3 volatile 关键字

volatile 解决了可见性问题，让修改成为原子操作。由于存在 happens-before 关系，在接下来读取 volatile 变量前，先对 volatile 变量进行写操作。 从而保证了对该字段的任何读操作都能督读到最近一次修改后的值。

class VolatileFlag implements Runnable {
  private volatile boolean shouldStop;
  public void run() {
    while (!shouldStop) {
      // 执行操作
    }
    System.out.println("Stopped.");
  }
  void stop() {
    shouldStop = true;
  }
  public static void main(String[] args) throws InterruptedException {
    VolatileFlag flag = new VolatileFlag();
    Thread thread = new Thread(flag);
    thread.start();
    flag.stop();
    thread.join();
  }
}

4.4 Atomic

java.util.concurrent.atomic package 包含了一组类，它们用类似 volatile 的无锁方式支持单个值的原子复合操作。

使用 AtomicXXX 类，可以实现 check-then-act 原子操作：

class CheckThenAct {
  private final AtomicReference value = new AtomicReference<>();
  void initialize() {
    if (value.compareAndSet(null, "Initialized value")) {
      System.out.println("Initialized only once.");
    }
  }
}

AtomicInteger 和 AtomicLong 都提供原子 increment/decrement 操作：

class Increment {
  private final AtomicInteger state = new AtomicInteger();
  void advance() {
    int oldState = state.getAndIncrement();
    System.out.println("Advanced: '" + oldState + "' -> '" + (oldState + 1) + "'.");
  }
}

如果你希望有这样一个计数器，不需要在获取计数的时候具备原子性，可以考虑用 LongAdder 取代 AtomicLong/AtomicInteger。 LongAdder 能在多个单元中存值并在需要时增加计数，因此在竞争激烈的情况下表现更好。

4.5 ThreadLocal

一种在线程中包含数据但不用锁的方法是使用 ThreadLocal 存储。从概念上讲，ThreadLocal 可以看做每个 Thread 存有一份自己的变量。Threadlocal 通常用于保存每个线程的值，比如“当前事务”或其他资源。 此外，还可以用于维护每个线程的计数器、统计信息或 ID 生成器。

class TransactionManager {
  private final ThreadLocal currentTransaction 
      = ThreadLocal.withInitial(NullTransaction::new);
  Transaction currentTransaction() {
    Transaction current = currentTransaction.get();
    if (current.isNull()) {
      current = new TransactionImpl();
      currentTransaction.set(current);
    }
    return current;
  }
}

5. 安全地发布对象

想让一个对象在当前作用域外使用可以发布对象，例如从 getter 返回该对象的引用。 要确保安全地发布对象，仅在对象完全构造好后发布，可能需要同步。 可以通过以下方式安全地发布：

• 静态初始化器。只有一个线程可以初始化静态变量，因为类的初始化在获取排他锁条件下完成。

class StaticInitializer {
  // 无需额外初始化条件，发布一个不可变对象
  public static final Year year = Year.of(2017); 
  public static final Set keywords;
  // 使用静态初始化器构造复杂对象
  static {
    // 创建可变集合
    Set keywordsSet = new HashSet<>(); 
    // 初始化状态
    keywordsSet.add("java");
    keywordsSet.add("concurrency");
    // 设置 set 不可修改 
    keywords = Collections.unmodifiableSet(keywordsSet); 
  }
}

• volatile 字段。由于写入 volatile 变量发生在读操作之前，因此读线程总能读到最新的值。

class Volatile {
  private volatile String state;
  void setState(String state) {
    this.state = state;
  }
  String getState() {
    return state; 
  }
}

• Atomic。例如 AtomicInteger 将值存储在 volatile 字段中，所以 volatile 变量的规则在这里也适用。

class Atomics {
  private final AtomicInteger state = new AtomicInteger();
  void initializeState(int state) {
    this.state.compareAndSet(0, state);
  }
  int getState() {
    return state.get();
  }
}

• final 字段

    

class Final {
  private final String state;
  Final(String state) {
    this.state = state;
  }
  String getState() {
    return state;
  }
}

确保在对象构造期间不会修改此引用。

class ThisEscapes {
 private final String name;
 ThisEscapes(String name) {
   Cache.putIntoCache(this);
   this.name = name;
 }
 String getName() { return name; }
}
class Cache {
 private static final Map CACHE = new ConcurrentHashMap<>();
 static void putIntoCache(ThisEscapes thisEscapes) {
   // 'this' 引用在对象完全构造之前发生了改变
   CACHE.putIfAbsent(thisEscapes.getName(), thisEscapes);
 }
}

• 正确同步字段

class Synchronization {
  private String state;
  synchronized String getState() {
    if (state == null)
      state = "Initial";
    return state;
  }
}

6. 不可变对象

不可变对象的一个重要特征是线程安全，因此不需要同步。要成为不可变对象：

• 所有字段都标记 final

• 所有字段必须是可变或不可变的对象，注意不要改变对象作用域，否则构造后不能改变对象状态

• this 引用在构造对象时不要泄露

• 类标记 final，子类无法重载改变类的行为

• 不可变对象示例：

// 标记为 final，禁止继承
public final class Artist {
  // 不可变变量，字段标记 final
  private final String name; 
  // 不可变变量集合, 字段标记 final
  private final List tracks; 
  public Artist(String name, List tracks) {
    this.name = name;
    // 防御性拷贝
    List copy = new ArrayList<>(tracks); 
    // 使可变集合不可修改
    this.tracks = Collections.unmodifiableList(copy); 
    // 构造对象期间，'this' 不传递到任何其他地方
  }
  // getter、equals、hashCode、toString 方法
}
// 标记为 final，禁止继承
public final class Track { 
  // 不可变变量，字段标记 final
  private final String title; 
  public Track(String title) {
    this.title = title;
  }
  // getter、equals、hashCode、toString 方法
}

7. 线程

java.lang.Thread 类用于表示应用程序线程或 JVM 线程。 代码始终在某个 Thread 类的上下文中执行，使用 Thread#currentThread() 可返回自己的当前线程。

表4 线程状态

表5 线程协调方法

7.1 如何处理 InterruptedException？

• 清理所有资源，并在当前运行级别尽可能能完成线程执行

• 当前方法声明抛出 InterruptedException。

• 如果方法没有声明抛出 InterruptedException，那么应该通过调用 Thread.currentThread().interrupt() 将中断标志恢复为 true。 并且在这个级别上抛出更合适的异常。为了能在更高调用级别上处理中断，把中断标志设置为 true 非常重要

7.2 处理意料之外的异常

线程可以指定一个 UncaughtExceptionHandler 接收由于发生未捕获异常导致线程突然终止的通知。

Thread thread = new Thread(runnable);
thread.setUncaughtExceptionHandler((failedThread, exception) -> {
  logger.error("Caught unexpected exception in thread '{}'.",
      failedThread.getName(), exception);
});
thread.start();

8. 活跃度

8.1 死锁

有多个线程，每个线程都在等待另一个线程持有的资源，形成一个获取资源的线程循环，这时会发生死锁。最典型的资源是对象 Monitor ，但也可能是任何可能导致阻塞的资源，例如 wait/notify。

下面的代码可能产生死锁：

class Account {
  private long amount;
  void plus(long amount) { this.amount += amount; }
  void minus(long amount) {
    if (this.amount < amount)
      throw new IllegalArgumentException();
    else
      this.amount -= amount;
  }
  static void transferWithDeadlock(long amount, Account first, Account second){
    synchronized (first) {
      synchronized (second) {
        first.minus(amount);
        second.plus(amount);
      }
    }
  }
}

如果同时出现以下情况，就会发生死锁：

• 一个线程正试图从第一个帐户切换到第二个帐户，并已获得了第一个帐户的锁

• 另一个线程正试图从第二个帐户切换到第一个帐户，并已获得第二个帐户的锁

避免死锁的方法：

• 按顺序加锁：总是以相同的顺序获取锁

class Account {
  private long id;
  private long amount;
  // 此处略去了一些方法
  static void transferWithLockOrdering(long amount, Account first, Account second){
    boolean lockOnFirstAccountFirst = first.id < second.id;
    Account firstLock = lockOnFirstAccountFirst  ? first  : second;
    Account secondLock = lockOnFirstAccountFirst ? second : first;
    synchronized (firstLock) {
      synchronized (secondLock) {
        first.minus(amount);
        second.plus(amount);
      }
    }
  }
}

• 锁定超时：获取锁时不要无限期阻塞，而是释放所有锁并重试

class Account {
  private long amount;
  // 此处略去了一些方法
  static void transferWithTimeout(
      long amount, Account first, Account second, int retries, long timeoutMillis
  ) throws InterruptedException {
    for (int attempt = 0; attempt < retries; attempt++) {
      if (first.lock.tryLock(timeoutMillis, TimeUnit.MILLISECONDS))
      {
        try {
          if (second.lock.tryLock(timeoutMillis, TimeUnit.MILLISECONDS))
          {
            try {
              first.minus(amount);
              second.plus(amount);
            }
            finally {
              second.lock.unlock();
            }
          }
        }
        finally {
          first.lock.unlock();
        }
      }
    }
  }
}

Jvm 能够检测 Monitor 死锁，并以线程转储的形式打印死锁信息。

8.2 活锁与线程饥饿

当线程将所有时间用于协商资源访问或者检测避免死锁，以至于没有线程能够访问资源时，会造成活锁（Livelock）。 线程饥饿发生在线程长时间持锁，导致一些线程无法继续执行被“饿死”。

9. java.util.concurrent

9.1 线程池

线程池的核心接口是 ExecutorService。 java.util.concurrent 还提供了一个静态工厂类 Executors，其中包含了新建线程池的工厂方法，新建的线程池参数采用最常见的配置。

表6 静态工厂方法

译注：在并行计算中，work-stealing 是一种针对多线程计算机程序的调度策略。 它解决了在具有固定数量处理器或内核的静态多线程计算机上执行动态多线程计算的问题，这种计算可以“产生”新的执行线程。 在执行时间、内存使用和处理器间通信方面都能够高效地完成任务。

在调整线程池的大小时，通常需要根据运行应用程序的计算机中的逻辑核心数量来确定线程池的大小。 在 Java 中，可以通过调用 Runtime.getRuntime().availableProcessors() 读取。

表7 线程池实现

可通过 ExecutorService#submit、ExecutorService#invokeAll 或 ExecutorService#invokeAny 提交任务，可根据不同任务进行多次重载。

表8 任务的功能接口

9.2 Future

Future 是对异步计算的一种抽象，代表计算结果。计算结果可能是某个计算值或异常。ExecutorService 的大多数方法都使用 Future 作为返回类型。使用 Future 时，可通过提供的接口检查当前状态，或者一直阻塞直到结果计算完成。

ExecutorService executorService = Executors.newSingleThreadExecutor();
Future future = executorService.submit(() -> "result");
try {
  String result = future.get(1L, TimeUnit.SECONDS);
  System.out.println("Result is '" + result + "'.");
} 
catch (InterruptedException e) {
  Thread.currentThread().interrupt();
  throw new RuntimeException(e);
} 
catch (ExecutionException e) {
  throw new RuntimeException(e.getCause());
} 
catch (TimeoutException e) {
  throw new RuntimeException(e);
}
assert future.isDone();

9.3 锁

9.3.1 Lock

java.util.concurrent.locks package 提供了标准 Lock 接口。ReentrantLock 在实现 synchronized 关键字功能的同时还包含了其他功能，例如获取锁的状态信息、非阻塞 tryLock() 和可中断锁定。直接使用 ReentrantLock 示例如下：

class Counter {
  private final Lock lock = new ReentrantLock();
  private int value;
  int increment() {
    lock.lock();
    try {
      return ++value;
    } finally {
      lock.unlock();
    }
  }
}

9.3.2 ReadWriteLock

java.util.concurrent.locks package 还包含 ReadWriteLock 接口（以及 Reentrantreadelock 实现）。该接口定义了一对锁进行读写操作，通常支持多个并发读取，但只允许一个写入。

class Statistic {
  private final ReadWriteLock lock = new ReentrantReadWriteLock();
  private int value;
  void increment() {
    lock.writeLock().lock();
    try {
      value++;
    } finally {
      lock.writeLock().unlock();
    }
  }
  int current() {
    lock.readLock().lock();
    try {
      return value;
    } finally {
      lock.readLock().unlock();
    }
  }
}

9.3.3 CountDownLatch

CountDownLatch 用一个计数器初始化。线程可以调用 await() 等待计数归零。其他线程（或同一线程）可能会调用 countDown() 来减小计数。一旦计数归零即不可重用。CountDownLatch 用于发生某些操作时触发一组未知的线程。

9.3.4 CompletableFuture

CompletableFuture 是对异步计算的一种抽象。 与普通 Future 不同，CompletableFuture 仅支持阻塞方式获得结果。当结果产生或发生异常时，执行由已注册的回调函数创建的任务管道。无论是创建过程中（通过 CompletableFuture#supplyAsync/runAsync），还是在加入回调过程中（*async 系列方法），如果没有指定标准的全局 ForkJoinPool#commonPool 都可以设置执行计算的执行器。

考虑 CompletableFuture 已执行完毕，那么通过非 *async 方法注册的回调将在调用者的线程中执行。

如果程序中有几个 future，可以使用 CompletableFuture#allOf 获得一个 future，这个 future 在所有 future 完成时结束。也可以调用 CompletableFuture#anyOf 获得一个 future，这个 future 在其中任何一个 future 完成时结束。

ExecutorService executor0 = Executors.newWorkStealingPool();
ExecutorService executor1 = Executors.newWorkStealingPool();
// 当这两个 future 完成时结束
CompletableFuture waitingForAll = CompletableFuture
    .allOf(
        CompletableFuture.supplyAsync(() -> "first"),
        CompletableFuture.supplyAsync(() -> "second", executor1)
    )
    .thenApply(ignored -> " is completed.");
CompletableFuture future = CompletableFuture.supplyAsync(() -> "Concurrency Refcard", executor0)
    // 使用同一个 executor
    .thenApply(result -> "Java " + result)
    // 使用不同的 executor
    .thenApplyAsync(result -> "Dzone " + result, executor1)
    // 当前与其他 future 完成后结束
    .thenCombine(waitingForAll, (first, second) -> first + second)
    // 默认使用 ForkJoinPool#commonPool 作为 executor
    .thenAcceptAsync(result -> {
      System.out.println("Result is '" + result + "'.");
    })
    // 通用处理
    .whenComplete((ignored, exception) -> {
      if (exception != null)
        exception.printStackTrace();
    });
// 第一个阻塞调用：在 future 完成前保持阻塞
future.join();
future
    // 在当前线程(main)中执行
    .thenRun(() -> System.out.println("Current thread is '" + Thread.currentThread().getName() + "'."))
    // 默认使用 ForkJoinPool#commonPool 作为 executor
    .thenRunAsync(() -> System.out.println("Current thread is '" + Thread.currentThread().getName() + "'."))

9.4 并发集合

使集合线程安全最简单方法是使用 Collections#synchronized* 系列方法。 由于这种解决方案在竞争激烈的情况下性能很差，所以 java.util.concurrent 提供了多种针对并发优化的数据结构。

9.4.1 List

表9：java.util.concurrent 中的 Lists

译注：copy-on-write（写入时复制）是一种计算机程序设计领域的优化策略。其核心思想是，如果有多个调用者同时请求相同资源（如内存或磁盘上的数据存储），他们会共同获取相同的指针指向相同的资源，直到某个调用者试图修改资源的内容时，系统才会真正复制一份专用副本给该调用者，而其他调用者所见到的最初的资源仍然保持不变。这个过程对其他的调用者透明。这种做法的主要优点是如果调用者没有修改该资源，就不会新建副本，因此多个调用者只是读取操作可以共享同一份资源。

9.4.2 Map

表10 java.util.concurrent 中的 Map

9.4.3 Set

表11 java.util.concurrent 中的 Set

封装 concurrent map 进而创建 concurrent set 的另一种方法：

Set concurrentSet = Collections.newSetFromMap(new ConcurrentHashMap());

9.4.4 Queue

队列就像是“生产者”和“消费者”之间的管道。按照“先进先出（FIFO）”顺序，将对象从管道的一端加入，从管道的另一端取出。BlockingQueue 接口继承了 Queue接口，并且增加了（生产者添加对象时）队列满或（消费者读取或移除对象时）队列空的处理。 在这些情况下，BlockingQueue 提供的方法可以一直保持或在一段时间内保持阻塞状态，直到等待的条件由另一个线程的操作改变。

表12 java.util.concurrent 中的 Queue

看完本文有收获？请转发分享给更多人

关注「ImportNew」，提升Java技能

喜欢就点「好看」呗~

以上所述就是小编给大家介绍的《Core Java 并发：理解并发概念》，希望对大家有所帮助，如果大家有任何疑问请给我留言，小编会及时回复大家的。在此也非常感谢大家对 码农网 的支持！

查看所有标签

本站部分资源来源于网络，本站转载出于传递更多信息之目的，版权归原作者或者来源机构所有，如转载稿涉及版权问题，请联系我们。