Concurrency in Java

一般地，下面两种方法可创建一个 Java 线程。
方法一：实现 Runnable 接口。它的写法有很多，如：

public class HelloRunnable implements Runnable {  // 直接实现Runnable接口

  public void run() {
    System.out.println("Hello from a thread!");
  }

  public static void main(String args[]) {
    Thread thread = new Thread(new HelloRunnable());
    thread.start();
  }
}

也可以这样：

public class HelloRunnable {
  public static void main(String[] args) {
    Thread thread = new Thread(new Runnable() {
      @Override
      public void run() {
        System.out.println("Hello from a thread!");
      }
    });
    thread.start();
  }
}

还可以用 Lambda Expressions：

public class HelloRunnable {
  public static void main(String[] args) {
    Thread thread = new Thread(() -> {   // 这是Lambda Expressions
      System.out.println("Hello from a thread!");
    });
    thread.start();
  }
}

方法二：扩展 Thread 类，重载其 run 方法：

public class HelloThread extends Thread {

  public void run() {
    System.out.println("Hello from a thread!");
  }

  public static void main(String args[]) {
    Thread thread = new HelloThread();
    thread.start();
  }
}

参考：
Defining and Starting a Thread: https://docs.oracle.com/javase/tutorial/essential/concurrency/runthread.html

A thread can be in one of the following states:

NEW

A thread that has not yet started is in this state.

RUNNABLE

A thread executing in the Java virtual machine is in this state. （处于 RUNABLE 状态的线程可能正在执行，也可能正在等待 CPU 为它分配执行时间）

BLOCKED

A thread that is blocked waiting for a monitor lock is in this state.

WAITING

A thread that is waiting indefinitely for another thread to perform a particular action is in this state. （处于 WAITING 状态的线程会一直等待，直到其他线程唤醒它，比如 Process 类的 waitFor()方法会使当前线程进入 WAITING 状态）

TIMED_WAITING

A thread that is waiting for another thread to perform an action for up to a specified waiting time is in this state. （过指定的一段时间后，会由系统自动唤醒，比如 Process 类的 waitFor(long timeout, TimeUnit unit)方法会使当前线程进入 TIMED_WAITING 状态）

TERMINATED

A thread that has exited is in this state.

参考：
https://docs.oracle.com/javase/7/docs/api/java/lang/Thread.State.html
http://www.uml-diagrams.org/examples/java-6-thread-state-machine-diagram-example.html

Java 从语言角度实现了 Monitor ，可以认为 Java 中每一个对象都有一个内部的锁(Intrinsic Lock)。

一般地，Intrinsic Lock（即使用 synchronized 关键字）可以满足大部分对锁的需求。

和 Intrinsic Lock 相比，ReentrantLock（在 Java SE 5.0 中引入）的功能更加丰富，比如：(1) ReentrantLock 支持 tryLock()操作；(2)ReentrantLock 支持“公平锁”。

用 ReentrantLock 保护代码块的基本结构如下：

private ReentrantLock myLock = new ReentrantLock();

public void fun1() {
    // some code
    myLock.lock();
    try {
        // critical section.
    } finally {
        myLock.unlock();         // 千万不要忘记unlock()。放入finally子句中，有异常抛出时也会执行unlock()。
    }
    // some code
}

上面结构确保任何时刻只有一个线程进入临界区。一旦一个线程得到了锁对象，其他任何线程都无法通过 lock 语句。当其他线程调用 lock 时，它们被阻塞，直到第一个线程释放锁对象。

说明 1：ReentrantLock 从其“名字”上看，叫“可重入锁”（注：Intrinsic Lock 也是可重入锁，“可重入锁”比“不可重入锁”更加安全）。“可重入锁”的意思是同一个线程可以重复地获得已经持有的锁。锁保持一个“持有计数”来跟踪对 lock 方法的嵌套调用。线程在每一次调用 lock 都要调用 unlock 来释放锁。所以， 被一个可重入锁保护的代码可以调用另一个使用相同锁的方法。
说明 2：pthread 的互斥锁默认为“不可重入的”（如果线程获得了锁，当这个线程再次获得同一个锁时会死锁），不过可以通过设置 PTHREAD_MUTEX_RECURSIVE 来使其变为可重入锁，详情可参考 man pthread_mutexattr_init 。

如果对数据进行“读”操作的线程远多于对其进行“写”操作的线程时，读/写锁可以提供更好的并发性。

下面是使用读/写锁的必要步骤：

1. Construct a ReentrantReadWriteLock object:
    private ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();

2. Extract the read and write locks:
    private Lock readLock = rwl.readLock();          // 多个读操作可以共用一个“读锁”，“读锁”会排斥写操作。
    private Lock writeLock = rwl.writeLock();        // “写锁”会排斥其他的读操作和写操作。

3. Use the read lock in all accessors:
    public double getTotalBalance()
    {
    readLock.lock();
    try { . . . }
    finally { readLock.unlock(); }
    }

4. Use the write lock in all mutators:
    public void transfer(. . .)
    {
    writeLock.lock();
    try { . . . }
    finally { writeLock.unlock(); }
    }

Java 8 中引入了 StampedLock ，提供了 writeLock() 和 readLock() 方法，可以用来代替 ReentrantReadWriteLock。除此外， StampedLock 提供了 tryOptimisticRead() 方法，用它可实现“乐观读”，当读操作多于写操作时（这往往是真实的场景），可以得到比 ReentrantReadWriteLock 更好的性能。

参考：Phaser And StampedLock Concurrency Synchronizers, by Dr Heinz M. Kabutz

volatile 仅能提供“很弱的”同步。相关知识点，请参考 Java 内存模型。

下面是一个实例展示了如何使用阻塞队列来控制一组线程。程序在一个目录及它的所有子目录中搜索所有文件，打印出包含指定关键字的行。

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.Scanner;

/**
 * @version 1.01 2012-01-26
 * @author Cay Horstmann
 */
public class BlockingQueueTest
{
  public static void main(String[] args)
  {
    Scanner in = new Scanner(System.in);
    System.out.print("Enter base directory (e.g. /usr/local/jdk1.6.0/src): ");
    String directory = in.nextLine();
    System.out.print("Enter keyword (e.g. volatile): ");
    String keyword = in.nextLine();

    final int FILE_QUEUE_SIZE = 10;
    final int SEARCH_THREADS = 100;

    BlockingQueue<File> queue = new
      ArrayBlockingQueue<>(FILE_QUEUE_SIZE);

    FileEnumerationTask enumerator = new
      FileEnumerationTask(queue, new File(directory));
    new Thread(enumerator).start();
    for (int i = 1; i <= SEARCH_THREADS; i++)
      new Thread(new SearchTask(queue, keyword)).start();
  }
}

/**
 * This task enumerates all files in a directory and its subdirectories.
 */
class FileEnumerationTask implements Runnable
{
  public static File DUMMY = new File("");
  private BlockingQueue<File> queue;
  private File startingDirectory;

  /**
   * Constructs a FileEnumerationTask.
   * @param queue the blocking queue to which the enumerated files are added
   * @param startingDirectory the directory in which to start the enumeration
   */
  public FileEnumerationTask(BlockingQueue<File> queue, File startingDirectory)
  {
    this.queue = queue;
    this.startingDirectory = startingDirectory;
  }

  public void run()
  {
    try {
      enumerate(startingDirectory);
      queue.put(DUMMY);
    } catch (InterruptedException e) {
    }
  }

  /**
   * Recursively enumerates all files in a given directory and its subdirectories.
   * @param directory the directory in which to start
   */
  public void enumerate(File directory) throws InterruptedException
  {
    File[] files = directory.listFiles();
    for (File file : files) {
      if (file.isDirectory()) enumerate(file);
      else queue.put(file);
    }
  }
}
/**
 * This task searches files for a given keyword.
 */
class SearchTask implements Runnable
{
  private BlockingQueue<File> queue;
  private String keyword;

  /**
   * Constructs a SearchTask.
   * @param queue the queue from which to take files
   * @param keyword the keyword to look for
   */
  public SearchTask(BlockingQueue<File> queue, String keyword)
  {
    this.queue = queue;
    this.keyword = keyword;
  }

  public void run()
  {
    try {
      boolean done = false;
      while (!done) {
        File file = queue.take();
        if (file == FileEnumerationTask.DUMMY) {
            queue.put(file);
            done = true;
        } else search(file);
      }
    } catch (IOException e) {
      e.printStackTrace();
    } catch (InterruptedException e) {
    }
  }

  /**
   * Searches a file for a given keyword and prints all matching lines.
   * @param file the file to search
   */
  public void search(File file) throws IOException
  {
    Scanner in = new Scanner(new FileInputStream(file));
    int lineNumber = 0;
    while (in.hasNextLine()) {
      lineNumber++;
      String line = in.nextLine();
      if (line.contains(keyword))
        System.out.printf("%s:%d:%s%n", file.getPath(), lineNumber, line);
    }
    in.close();
  }
}

在上面程序中，生产者线程枚举在所在子目录下的所有文件并把它们放到一个阻塞队列中。同时启动了大量搜索线程。每个搜索线程从队列中取出一个文件，打开它，打印所有包含该关键字的行，然后取出下一个文件。在这个程序中我们使用了一个小技巧在工作结束后终止这个应用程序：为了发出完成信号，生产者线程放置一个虚拟对象（DUMMY）到队列中（这就像行李传送带上放一个写着“最后一个包”的虚拟包）。当搜索线程取到这个虚拟对象时，将其放回并终止。

注意：这里不需要显式的线程同步。在这个应用程序中，我们使用阻塞队列作为一种同步机制。

死锁是指两个或两个以上的线程在执行过程中，因争夺资源而造成的一种互相等待的现象。

饥饿(Starvation)描述了一个线程由于得不到足够的资源而不能执行程序的现象。 当共享资源由一个“贪婪”的线程长时间占用时，其它线程可能出现“饥饿”。 例如，假设一个对象有一个同步的方法，需要很长时间才能返回。如果一个线程经常执行这个方法，其他需要同样访问这个资源的线程就有可能出现阻塞的情况。

活锁(Livelock)是指线程 1 可以使用资源，但它很礼貌，让其他线程先使用资源；而线程 2 也可以使用资源，但它也很礼貌，也让其他线程先使用资源。这样你让我，我让你，最后两个线程都无法使用资源。例如，马路中间有条小桥，只能容纳一辆车经过，桥两头开来两辆车 A 和 B，A 比较礼貌，示意 B 先过，B也比较礼貌，示意 A 先过，结果两人一直谦让谁也过不去。

参考：
Java - Thread Livelock: http://www.logicbig.com/tutorials/core-java-tutorial/java-multi-threading/thread-livelock/
Starvation and Livelock: http://docs.oracle.com/javase/tutorial/essential/concurrency/starvelive.html
《现代操作系统（原书第 3 版）》，6.7.3 活锁，6.7.4 饥饿

FutureTask 包装器将 Callable 转换成 Future 和 Runnable，它同时实现二者的接口。 这样，我们可以方便地指定一个 Callable 对象给新线程了（我们知道“用 new Thread 创建线程时，只能指定 Runnable 对象”，有了 FutureTask 包装器，就可以指定 Callable 对象给新线程；后文将说明如何使用线程池，由于线程池可以直接提交 Callable 对象，所以使用线程池时一般不需要 FutureTask 包装器）。

FutureTask 使用方法大致如下：

Callable<Integer> myComputation = . . .;
FutureTask<Integer> task = new FutureTask<Integer>(myComputation);
Thread t = new Thread(task);         // it's a Runnable
t.start();
...
Integer result = task.get();         // it's a Future

下面例子将计算目录及其所有子目录中包含某关键字的文件个数。

import java.io.*;
import java.util.*;
import java.util.concurrent.*;
/**
 * @version 1.01 2012-01-26
 * @author Cay Horstmann
 */
public class FutureTest
{
  public static void main(String[] args)
  {
    Scanner in = new Scanner(System.in);
    System.out.print("Enter base directory (e.g. /usr/local/jdk5.0/src): ");
    String directory = in.nextLine();
    System.out.print("Enter keyword (e.g. volatile): ");
    String keyword = in.nextLine();

    MatchCounter counter = new MatchCounter(new File(directory), keyword);
    FutureTask<Integer> task = new FutureTask<>(counter);
    Thread t = new Thread(task);
    t.start();
    try {
      System.out.println(task.get() + " matching files.");
    } catch (ExecutionException e) {
      e.printStackTrace();
    } catch (InterruptedException e) {
    }
  }
}

/**
 * This task counts the files in a directory and its subdirectories that contain a given keyword.
 */
class MatchCounter implements Callable<Integer>
{
  private File directory;
  private String keyword;
  private int count;

  /**
   * Constructs a MatchCounter.
   * @param directory the directory in which to start the search
   * @param keyword the keyword to look for
   */
  public MatchCounter(File directory, String keyword)
  {
    this.directory = directory;
    this.keyword = keyword;
  }

  public Integer call()
  {
    count = 0;
    try {
      File[] files = directory.listFiles();
      List<Future<Integer>> results = new ArrayList<>();

      for (File file : files)
        if (file.isDirectory()) {
          // 对于每一个子目录，产生一个新的MatchCounter并为它启动一个线程。
          MatchCounter counter = new MatchCounter(file, keyword);
          FutureTask<Integer> task = new FutureTask<>(counter);
          results.add(task);
          Thread t = new Thread(task);
          t.start();
        } else {
          if (search(file)) count++;
        }

      for (Future<Integer> result : results)
        try {
          count += result.get();         // 每一次对get的调用都会发生阻塞直到对应的call返回，结果可获得为止。
                                         // 当然线程是并行运行的，因此，很可能在大致相同的时刻所有结果都可获得。
        } catch (ExecutionException e) {
          e.printStackTrace();
        }
    } catch (InterruptedException e) {
    }
    return count;
  }

  /**
   * Searches a file for a given keyword.
   * @param file the file to search
   * @return true if the keyword is contained in the file
   */
  public boolean search(File file)
  {
    try {
      Scanner in = new Scanner(file);
      boolean found = false;
      while (!found && in.hasNextLine()) {
        String line = in.nextLine();
        if (line.contains(keyword)) found = true;
      }
      in.close();
      return found;
    } catch (IOException e) {
      return false;
    }
  }
}

说明：这个实例不是很好，因为它为每个目录产生一个线程，这样可能产生大量的生命期很短的线程。可以用“线程池”来避免大量生命期很短的线程，后文将介绍如何用“线程池”实现和这个实现相同的功能。

Futures and promises are pretty similar concepts, the difference is that a future is a read-only container for a result that does not yet exist, while a promise can be written (normally only once). The Java 8 CompletableFuture can be thought of as promises.

参考：http://stackoverflow.com/questions/14541975/difference-between-future-and-promise

通过 newCachedThreadPool/newFixedThreadPool/newSingleThreadExecutor 创建了线程池后，可以用下面的 submit 方法之一将一个 Runnable 对象或者 Callable 对象提交给 ExecutorService:

// 使用submit方法可向线程池提交任务，线程池会在方便的时候尽早地安排线程来执行提交的任务。
Future<?> submit(Runnable task)            // 对返回的Future对象调用get方法时，会在完成的时候简单地返回null
Future<T> submit(Runnable task, T result)  // 对返回的Future对象调用get方法时，会在完成的时候返回result
Future<T> submit(Callable<T> task)         // 对返回的Future对象调用get方法时，会在完成的时候返回Callable对象的call方法的返回值

该线程池会在方便的时候尽早地执行提交的任务。调用 submit 时，会得到一个 Future 对象，可用来查询该任务的状态。

当不再需要提交任务时，可以调用 shutdown 来关闭线程池。被关闭的线程池不再接受新的任务，当所有任务都完成后，线程池中线程会死亡。另一种方法是调用 shutdownNow ，它会取消尚未开始的所有任务并试图中断正在运行的线程。

说明：用 newScheduledThreadPool/newSingleThreadScheduledExecutor 创建的线程池，通过调用 schedule 等方法可以在预定时间执行任务。

下面是一个 Callable, Future, Executor 的用法举例。

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class MyCallable implements Callable<String> {

    @Override
    public String call() throws Exception {
        // 实现业务逻辑，这里简单地返回线程名字
        Thread.sleep(1000);
        return Thread.currentThread().getName();
    }

    public static void main(String args[]) {
        ExecutorService executor = Executors.newFixedThreadPool(3);   // 创建包含3个线程的线程池。

        List<Future<String>> list = new ArrayList<Future<String>>();

        // Create MyCallable instance
        Callable<String> callable = new MyCallable();
        for(int i=0; i< 10; i++) {
            Future<String> future = executor.submit(callable);        // 向线程池提交任务。
            list.add(future);            // 把future保存到列表中，以便后面引用。
        }

        for(Future<String> fut : list) {
            try {
                // print the return value of Future, notice the output delay in console
                // because Future.get() waits for task to get completed
                String ret = fut.get();   // fut.get()会阻塞直到对应call方法返回，且call方法的返回值就是get()返回值。
                System.out.println(ret);
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
        }

        executor.shutdown();
    }
}

上面程序的测试输出如下（你可能得到不一样的结果）：

pool-1-thread-1
pool-1-thread-2
pool-1-thread-3
pool-1-thread-1
pool-1-thread-3
pool-1-thread-2
pool-1-thread-2
pool-1-thread-1
pool-1-thread-3
pool-1-thread-3

这个例子和前面介绍的例子功能相同（即计算目录及其所有子目录中包含某关键字的文件个数），只是用线程池实现。

import java.io.*;
import java.util.*;
import java.util.concurrent.*;

/**
 * @version 1.01 2012-01-26
 * @author Cay Horstmann
 */
public class ThreadPoolTest
{
  public static void main(String[] args) throws Exception
  {
    Scanner in = new Scanner(System.in);
    System.out.print("Enter base directory (e.g. /usr/local/jdk5.0/src): ");
    String directory = in.nextLine();
    System.out.print("Enter keyword (e.g. volatile): ");
    String keyword = in.nextLine();

    ExecutorService pool = Executors.newCachedThreadPool();
    MatchCounter counter = new MatchCounter(new File(directory), keyword, pool);
    Future<Integer> result = pool.submit(counter);

    try {
      System.out.println(result.get() + " matching files.");
    } catch (ExecutionException e) {
      e.printStackTrace();
    } catch (InterruptedException e) {
    }
    pool.shutdown();

    int largestPoolSize = ((ThreadPoolExecutor) pool).getLargestPoolSize();
    System.out.println("largest pool size=" + largestPoolSize);
  }
}

/**
 * This task counts the files in a directory and its subdirectories that contain a given keyword.
 */
class MatchCounter implements Callable<Integer>
{
  private File directory;
  private String keyword;
  private ExecutorService pool;
  private int count;

  /**
   * Constructs a MatchCounter.
   * @param directory the directory in which to start the search
   * @param keyword the keyword to look for
   * @param pool the thread pool for submitting subtasks
   */
  public MatchCounter(File directory, String keyword, ExecutorService pool)
  {
    this.directory = directory;
    this.keyword = keyword;
    this.pool = pool;
  }

  public Integer call()
  {
    count = 0;
    try {
      File[] files = directory.listFiles();
      List<Future<Integer>> results = new ArrayList<>();

      for (File file : files) {
        if (file.isDirectory()) {
          MatchCounter counter = new MatchCounter(file, keyword, pool);
          Future<Integer> result = pool.submit(counter);
          results.add(result);
        }
        else {
          if (search(file)) count++;
        }
      }

      for (Future<Integer> result : results) {
        try {
          count += result.get();
        } catch (ExecutionException e) {
          e.printStackTrace();
        }
      }
    }
    catch (InterruptedException e) {
    }
    return count;
  }

  /**
   * Searches a file for a given keyword.
   * @param file the file to search
   * @return true if the keyword is contained in the file
   */
  public boolean search(File file)
  {
    try {
      Scanner in = new Scanner(file);
      boolean found = false;
      while (!found && in.hasNextLine()) {
        String line = in.nextLine();
        if (line.contains(keyword)) found = true;
      }
      in.close();
      return found;
    } catch (IOException e) {
      return false;
    }
  }
}

The fork/join framework is an implementation of the ExecutorService interface that helps you take advantage of multiple processors. It is designed for work that can be broken into smaller pieces recursively. The goal is to use all the available processing power to enhance the performance of your application.

Fork/Join 是 Java 7 中提供的一个用于并行执行任务的框架，它是一个“把大任务分割成若干个小任务，最终汇总每个小任务结果后得到大任务结果”的框架。

下面是使用 Fork/Join 框架的基本步骤：
第一步，定义一个 ForkJoinTask（它是 Future 接口的一个实现，同时还增加了其它的方法）任务，它提供在任务中执行 fork()和 join()操作的机制，通常情况下我们不需要直接继承 ForkJoinTask 类，而只需要继承它的子类，Fork/Join 框架提供了以下两个常用子类：(1) RecursiveAction：用于没有返回结果的任务（即不关心任务的返回值）；(2) RecursiveTask：用于有返回结果的任务。
在定义自己的 ForkJoinTask 时，你需要实现 compute()方法，它用来定义如何分割以及合并任务。一般地，compute()方法的实现类似于下面的伪代码：

if (my portion of the work is small enough)
  do the work directly
else
  split my work into two pieces                   // fork 步骤
  invoke the two pieces and wait for the results  // join 步骤

第二步，通过 ForkJoinPool（它是 ExecutorService 接口的一个实现）提交前面定义的任务即可。ForkJoinPool 中有三种方法来提交 ForkJoinTask 任务：

public void execute(ForkJoinTask<?> task)                // 提交任务（异步地）。
public <T> T invoke(ForkJoinTask<T> task)                // 执行任务，且等待执行完成。返回任务的compute()方法的返回值。
public <T> ForkJoinTask<T> submit(ForkJoinTask<T> task)  // 提交任务（异步地）。返回ForkJoinTask，在返回的ForkJoinTask上调用get()方法可等待任务结束并得到任务的compute()方法的返回值。

参考：
Oracle Fork/Join Tutorial
Fork and Join: Java Can Excel at Painless Parallel Programming Too!

一个倒计时门栓让一个线程集等待直到计数变为 0。倒计时门栓是一次性的，一旦计数为 0，就不能再重用了。调用 countDownLatch.countDown()可以减少门栓的一个计数；调用 countDownLatch.await()时线程会等待，直到 countDownLatch 中的计数到 0 之后就会恢复执行。

下面是使用两个倒计时门栓用作线程同步的一个实例（摘自：https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/CountDownLatch.html ）：
Here is a pair of classes in which a group of worker threads use two countdown latches:

1. The first is a start signal that prevents any worker from proceeding until the driver is ready for them to proceed;
   
2. The second is a completion signal that allows the driver to wait until all workers have completed.
   

import java.util.concurrent.CountDownLatch;

class Driver {
    private static final int N = 10;

    public static void main(String args[]) throws InterruptedException {
        CountDownLatch startSignal = new CountDownLatch(1);
        CountDownLatch doneSignal = new CountDownLatch(N);

        for (int i = 0; i < N; ++i) // create and start threads
            new Thread(new Worker(startSignal, doneSignal)).start();

        doSomethingElse1();       // don't let run yet
        startSignal.countDown();  // let all threads proceed
        doSomethingElse2();
        doneSignal.await();       // wait for all to finish
        doSomethingElse3();
    }

    private static void doSomethingElse1() {  }
    private static void doSomethingElse2() {  }
    private static void doSomethingElse3() {  }
}

class Worker implements Runnable {
    private final CountDownLatch startSignal;
    private final CountDownLatch doneSignal;

    Worker(CountDownLatch startSignal, CountDownLatch doneSignal) {
        this.startSignal = startSignal;
        this.doneSignal = doneSignal;
    }

    public void run() {
        try {
            startSignal.await();        // 阻止线程运行，直到倒计时门栓startSignal的计数为零时，才会恢复运行。
            doWork();
            doneSignal.countDown();
        } catch (InterruptedException ex) {
        }
    }

    void doWork() {  }
}

CyclicBarrier 是一个同步辅助类，它可以使一组线程互相等待，直到到达某个公共屏障点(common barrier point)后又开始运行。

下面是如何使用它的一些细节。首先，构造一个循环屏障，并给出参与的线程数：

CyclicBarrier barrier = new CyclicBarrier(nthreads);

每一个线程完成一些准备工作后，在循环屏障上调用 await：

public void run() {
  doWork();
  barrier.await();
  // ...
}

这样当所有线程都准备完毕（调用 await 后），才又开始运行。

Exchanger 类允许在 2 个线程间定义同步点，当 2 个线程到达这个点，他们相互交换数据类型，使用第一个线程的数据类型变成第二个的，然后第二个线程的数据类型变成第一个的。

具体来说， 当线程 A 调用 Exchange 对象的 exchange()方法后，它会陷入阻塞状态，直到线程 B 也调用了 exchange()方法，然后以线程安全的方式交换数据，之后线程 A 和 B 继续运行。

下面是 Exchanger 的一个实例（摘自：http://www.javacodex.com/Concurrency/Exchanger-Example ）：

import java.util.concurrent.*;

public class ExchangerExample {

    public static void main(String[] args) {

        Exchanger<String> exchanger = new Exchanger<String>();

        Thread t1 = new MyThread(exchanger, "I like coffee.");
        Thread t2 = new MyThread(exchanger, "I like tea");
        t1.start();
        t2.start();
    }
}

class MyThread extends Thread {

    Exchanger<String> exchanger;
    String message;

    MyThread(Exchanger<String> exchanger, String message) {
        this.exchanger = exchanger;
        this.message = message;
    }

    public void run() {
        try {
            System.out.println(this.getName() + " message: " + message);

            // exchange messages
            message = exchanger.exchange(message);

            System.out.println(this.getName() + " message: " + message);
        } catch (Exception e) {
        }
    }
}

运行上面程序的输出（你可能得到顺序不一样的输出）为：

Thread-1 message: I like tea
Thread-0 message: I like coffee.
Thread-0 message: I like tea
Thread-1 message: I like coffee.

原子变量(Atomic Variables)和 volatile 变量有相同的“happens-before”原则，也就是说对一个原子变量的“写操作”会“先行发生”于后面对这个原子变量的“读操作”。原子变量和 volatile 变量相比的优点在于原子变量支持原子操作（比如 AtomicInteger 类有原子的 compareAndSet 等方法），而 volatile 变量不能直接提供类似 compareAndSet 方法的原子性保证。

一般来说，原子变量比锁要更加轻量级。原子变量是实现高性能并发，非阻塞同步的基本工具。

下面是不使用原子变量实现多线程安全的计数器：

// 多线程安全计数器
class SynchronizedCounter {
    private int c = 0;

    public synchronized void increment() {
        c++;
    }

    public synchronized void decrement() {
        c--;
    }

    public synchronized int value() {
        return c;
    }
}

用原子变量实现的多线程安全计数器：

// 用原子变量实现的多线程安全计数器
import java.util.concurrent.atomic.AtomicInteger;

class AtomicCounter {
    private AtomicInteger c = new AtomicInteger(0);

    public void increment() {
        c.incrementAndGet();
    }

    public void decrement() {
        c.decrementAndGet();
    }

    public int value() {
        return c.get();
    }
}

互斥同步（Mutual Exclusion＆Synchronization）是常见的一种并发正确性保障手段。同步是指在多个线程并发访问共享数据时，保证共享数据在同一个时刻只被一个（或者是一些，使用信号量的时候）线程使用。而互斥是实现同步的一种手段，临界区（Critical Section）、互斥量（Mutex）和信号量（Semaphore）都是主要的互斥实现方式。因此，在这 4 个字里面，互斥是因，同步是果；互斥是方法，同步是目的。

互斥同步最主要的问题就是进行线程阻塞和唤醒所带来的性能问题，因此这种同步也称为阻塞同步（Blocking Synchronization）。从处理问题的方式上说， 互斥同步属于一种悲观的并发策略 ， 总是认为只要不去做正确的同步措施（例如加锁），那就肯定会出现问题，无论共享数据是否真的会出现竞争，它都要进行加锁（这里讨论的是概念模型，实际上虚拟机会优化掉很大一部分不必要的加锁）、用户态核心态转换、维护锁计数器和检查是否有被阻塞的线程需要唤醒等操作。随着硬件指令集的发展，我们有了另外一个选择：基于冲突检测的乐观并发策略，通俗地说，就是先进行操作，如果没有其他线程争用共享数据，那操作就成功了； 如果共享数据有争用，产生了冲突，那就再采取其他的补偿措施（最常见的补偿措施就是不断地重试，直到成功为止），这种乐观的并发策略的许多实现都不需要把线程挂起，因此这种同步操作称为非阻塞同步（Non-Blocking Synchronization）。

为什么说使用乐观并发策略需要“硬件指令集的发展”才能进行呢？ 因为我们需要操作和冲突检测这两个步骤具备原子性，靠什么来保证呢？如果这里再使用互斥同步来保证就失去意义了，所以我们只能靠硬件来完成这件事情，硬件保证一个从语义上看起来需要多次操作的行为只通过一条处理器指令就能完成，这类指令常用的有：

• 测试并设置（Test-and-Set）。
  
• 获取并增加（Fetch-and-Increment）。
  
• 交换（Swap）。
  
• 比较并交换（Compare-and-Swap，下文称 CAS）。
  
• 加载链接/条件存储（Load-Linked/Store-Conditional，下文称 LL/SC）。
  

其中，前面的 3 条是 20 世纪就已经存在于大多数指令集之中的处理器指令，后面的两条是现代处理器新增的，而且这两条指令的目的和功能是类似的。在 IA64、x86 指令集中有 cmpxchg 指令完成 CAS 功能，在 sparc-TSO 也有 casa 指令实现，而在 ARM 和 PowerPC 架构下，则需要使用一对 ldrex/strex 指令来完成 LL/SC 的功能。

参考：深入理解 Java 虚拟机——JVM 高级特性与最佳实践(第 2 版)，13.2.2 线程安全的实现方法

伪随机数生成器（Pseudorandom Number Generator, PRNG）是确定的，完成由 seed 值决定， 相同的 seed 值总可以得到相同的随机数序列。

由于每次都生成相同的序列，所以 PRNG 一定是有状态的（它需要保存上一个产生的随机数，从而才知道下一个应该产生哪个伪随机数，从而每次重新运行可得到相同的序列）。

在 Java 中，已经实现了 PRNG（Random 类的方法 nextInt），下面是一个使用例子：

import java.util.Random;

public class RandomTest {
    public static void main(String args[]) {

        // create random object
        long seed = 12345;
        Random randomno = new Random(seed);

        // check next int value
        for (int i = 0; i < 5; i++) {
            System.out.println("Next int value: " + randomno.nextInt(10000));
        }
    }
}

上面例子，由于 seed 值是不变的，所以它的每次输出结果都会是一样的。通过测试可知，它的输出总是：

Next int value: 6251
Next int value: 9080
Next int value: 8241
Next int value: 8828
Next int value: 55

如何实现产生下一个伪随机数的方法 nextInt() 呢？最流行的算法是线性同余法，具体算法可参考：《计算机程序设计艺术——第 2 卷 半数值算法(第 3 版)，第 3 章随机数，3.2 节生成一致随机数》。本文的重点不是产生随机序列的具体算法，而是如何让它变得线程安全（即实现“线程安全”的 nextInt）。