本文是我们学院课程中名为Java Concurrency Essentials的一部分 。

在本课程中，您将深入探讨并发的魔力。 将向您介绍并发和并发代码的基础知识，并学习诸如原子性，同步和线程安全之类的概念。 在这里查看 ！

1.简介

下一章介绍java.util.concurrent包。 在该程序包中包含许多有趣的类，这些类提供了实现多线程应用程序所需的必要和有用的功能。 在讨论了如何使用Executor接口及其实现之后，本章介绍了原子数据类型和并发数据结构。 最后一部分向信号灯和倒数锁存器发出信号。

2. java.util.concurrent

阅读了先前关于并发和多线程的文章之后，您可能会觉得编写在多线程环境中执行良好的健壮代码并不总是那么简单。 有一个谚语可以说明这一点（来源未知）：

• 初级程序员认为并发很难。
• 经验丰富的程序员认为并发很容易。
• 高级程序员认为并发很难。

因此，一个可靠的数据结构和类库可提供经过良好测试的线程安全性，对于编写使用并发性程序的任何人都非常有帮助。 幸运的是，JDK为此目的提供了一组现成的数据结构和功能。 所有这些类都位于包java.util.concurrent中。

执行者

java.util.concurrent包定义了一组接口，这些接口的实现执行任务。 其中最简单的一个是Executor接口：

public interface Executor {void execute(Runnable command);
}

因此，执行器实现采用给定的Runnable实例并执行它。 该接口不对执行方式进行任何假设，javadoc仅声明“将来某个时候执行给定命令”。 因此，一个简单的实现可以是：

public class MyExecutor implements Executor {public void execute(Runnable r) {(new Thread(r)).start();}
}

除了纯接口外，JDK还提供了一个成熟且可扩展的实现，名为ThreadPoolExecutor 。 在后台， ThreadPoolExecutor维护线程池，并在给定execute()方法的情况下将Runnable实例调度到该池。 传递给构造函数的参数控制线程池的行为。 参数最多的构造函数如下：

ThreadPoolExecutor（int corePoolSize，int maximumPoolSize，long keepAliveTime，TimeUnit单位，BlockingQueue <Runnable> workQueue，ThreadFactory threadFactory，RejectedExecutionHandler处理程序）

让我们逐步研究不同的参数：

• corePoolSize ： ThreadPoolExecutor具有一个corePoolSize属性，该属性确定只有在队列已满时才启动新线程，直到启动新线程为止。
• maximumPoolSize ：此属性确定最大启动多少线程。 您可以将其设置为Integer.MAX_VALUE ，以使其没有上限。
• keepAliveTime ：当ThreadPoolExecutor创建的ThreadPoolExecutor数超过corePoolSize ，当线程在给定的时间内空闲时，该线程将从池中删除。
• unit ：这只是keepAliveTime的TimeUnit 。
• workQueue ：此队列保存通过execute()方法给定的Runnable实例，直到它们实际启动为止。
• threadFactory ：此接口的实现使您可以控制ThreadPoolExecutor使用的线程的创建。
• handler ：当您为workQueue指定固定大小并提供maximumPoolSize时，可能会发生ThreadPoolExecutor由于饱和而无法执行您的Runnable实例的情况。 在这种情况下，将调用提供的处理程序，并让您控制在这种情况下应该发生的情况。

由于有许多参数需要调整，让我们检查一些使用它们的代码：

public class ThreadPoolExecutorExample implements Runnable {private static AtomicInteger counter = new AtomicInteger();private final int taskId;public int getTaskId() {return taskId;}public ThreadPoolExecutorExample(int taskId) {this.taskId = taskId;}public void run() {try {Thread.sleep(5000);} catch (InterruptedException e) {e.printStackTrace();}}public static void main(String[] args) {BlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>(10);ThreadFactory threadFactory = new ThreadFactory() {public Thread newThread(Runnable r) {int currentCount = counter.getAndIncrement();System.out.println("Creating new thread: " + currentCount);return new Thread(r, "mythread" + currentCount);}};RejectedExecutionHandler rejectedHandler = new RejectedExecutionHandler() {public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {if (r instanceof ThreadPoolExecutorExample) {ThreadPoolExecutorExample example = (ThreadPoolExecutorExample) r;System.out.println("Rejecting task with id " + example.getTaskId());}}};ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 10, 1, TimeUnit.SECONDS, queue, threadFactory, rejectedHandler);for (int i = 0; i < 100; i++) {executor.execute(new ThreadPoolExecutorExample(i));}executor.shutdown();}
}

我们的run()实现仅睡5秒钟，但这不是此代码的主要重点。 ThreadPoolExecutor从5个核心线程开始，并允许池最多扩展到10个线程。 出于演示目的，我们仅将未使用的线程闲置大约1秒钟。 这里的队列实现是LinkedBlockingQueue与10个的容量Runnable实例。 我们还实现了一个简单的ThreadFactory以便跟踪线程的创建。 对于RejectedExecutionHandler也是如此。

在环路main()方法现在发出100 Runnable实例很短的时间量内该池。 该示例的输出显示我们必须创建10个线程（最多）来处理所有未决的Runnables ：

Creating new thread: 0
...
Creating new thread: 9
Rejecting task with id 20
...
Rejecting task with id 99

但它也显示所有taskId大于19的任务都转发到RejectedExecutionHandler 。 这是因为我们的Runnable实现休眠了5秒钟。 第10个线程已经启动后，队列只能持有另外10个Runnable实例。 然后，必须拒绝所有其他实例。

最后， shutdown()方法使ThreadPoolExecutor拒绝所有其他任务，并等待直到已执行的任务已执行。 您可以将调用shutdown()替换为shutdownNow() 。 后者尝试中断所有正在运行的线程并关闭线程池，而不等待所有线程完成。 在上面的示例中，您会看到十个InterruptedException异常，因为我们的十个睡眠线程被立即唤醒。

执行器服务

Executor接口非常简单，它仅强制底层实现实现execute()方法。 ExecutorService进一步扩展了Executor接口，并添加了一系列实用程序方法（例如，添加了完整的任务集合），关闭线程池的方法以及查询实现以获取执行结果的能力执行一项任务。 我们已经看到， Runnable接口仅定义一个run()方法作为返回值是无效的。 因此，有必要引入一个名为Callable的新接口，该接口类似于Runnable定义也只有一个方法，但是此方法返回一个值：

V call();

但是，JDK如何处理任务返回一个值但提交给线程池以执行的事实呢？

任务的提交者无法提前知道任务何时执行以及执行的持续时间。 让当前线程等待结果显然不是解决方案。 在另一个类java.util.concurrent.Future<V>实现了检查结果是否已经可用的功能，该功能可以阻止或等待一定时间。 此类只有几种方法可以检查任务是否已完成，取消任务以及检索其结果。

最后但并非最不重要的一点是，我们还有另一个接口，该接口通过某些方法扩展了Executor接口和ExecutorService接口，以在给定的时间点计划任务。 接口的名称为ScheduledExecutorService ，它基本上提供了一个schedule()方法，该方法使用一个参数来等待任务执行之前需要等待多长时间：

schedule(Callable<V> callable, long delay, TimeUnit unit);
schedule(Runnable command, long delay, TimeUnit unit);

就像ExecutorService一样， schedule()方法有两种变体：一种用于Runnable接口，一种用于使用Callable接口返回值的任务。 ScheduledExecutorService还提供了一种定期执行任务的方法：

scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit);

在初始延迟旁边，我们可以指定任务应运行的时间。

最后一个示例已经展示了如何创建ThreadPoolExecutor 。 ScheduledExecutorService的实现名为ScheduledThreadPoolExecutor ，其处理方式与上面使用的ThreadPoolExecutor非常相似。 但是通常不需要完全控制ExecutorService的所有功能。 试想一下，一个简单的测试客户端应该使用一个简单的ThreadPool调用一些服务器方法。

因此，JDK的创建者创建了一个名为Executors的简单工厂类（请注意结尾的）。 此类提供了一些静态方法来创建可使用的ThreadPoolExecutor 。 所有这些使我们能够实现一个简单的线程池，该线程池执行一堆计算一些数字的任务（这里的数字运算操作是出于演示目的，由一个简单的Thread.sleep()代替）：

public class ExecutorsExample implements Callable<Integer> {private static Random random = new Random(System.currentTimeMillis());public Integer call() throws Exception {Thread.sleep(1000);return random.nextInt(100);}public static void main(String[] args) throws InterruptedException, ExecutionException {ExecutorService executorService = Executors.newFixedThreadPool(5);Future<Integer>[] futures = new Future[5];for (int i = 0; i < futures.length; i++) {futures[i] = executorService.submit(new ExecutorsExample());}for (int i = 0; i < futures.length; i++) {Integer retVal = futures[i].get();System.out.println(retVal);}executorService.shutdown();}
}

ExecutorService的创建是ExecutorService 。 要执行一些任务，我们只需要一个for循环，即可创建ExecutorsExample的一些新实例并将返回的Future存储在数组中。 将任务提交给服务后，我们只需等待结果。 Future get()方法正在阻塞，即当前线程进入睡眠状态直到结果可用。 如果任务未在定义的时间段内完成，则此方法的重写版本采用超时规范，以便等待线程继续进行。

并发集合

Java集合框架包含每个Java程序员在日常工作中使用的各种数据结构。 此集合由java.util.concurrent包中的数据结构扩展。 这些实现提供了在多线程环境中使用的线程安全集合。

许多Java程序员甚至不知不觉地使用线程安全的数据结构。 “旧”类Hashtable和Vector是此类的示例。 自1.0版以来，它们是JDK的一部分，这些基本数据结构在设计时考虑了线程安全性。 尽管此处的线程安全性仅意味着所有方法都在实例级别上同步。 以下代码取自Oracle的JDK实现：

public synchronized void clear() {Entry tab[] = table;modCount++;for (int index = tab.length; --index >= 0; )tab[index] = null;count = 0;
}

这与诸如HashMap或ArrayList （自JDK 1.2起都提供）之类的“较新”集合类（它们本身都不是线程安全的）的关键区别。 但是，有一种方便的方法可以检索此类“较新”的集合类的线程安全实例：

HashMap<Long,String> map = new HashMap<Long, String>();
Map<Long, String> synchronizedMap = Collections.synchronizedMap(map);

正如我们在上面的代码中看到的那样， Collections类使我们可以在运行时创建以前未同步的collections类的同步版本。

如前所述，将关键字sync同步到方法会导致在每个时间点只有一个线程执行所研究对象的方法。 当然，这是使简单集合类具有线程安全性的最简单方法。 更高级的技术包括专为并发访问而设计的特殊算法。 这些算法在java.util.concurrent包的集合类java.util.concurrent实现。

此类的一个示例是ConcurrentHashMap ：

ConcurrentHashMap<Long,String> map = new ConcurrentHashMap<Long,String>();
map.put(key, value);
String value2 = map.get(key);

上面的代码看起来与普通的HashMap几乎相同，但是底层实现却完全不同。 ConcurrentHashMap不是将整个表仅使用一个锁，而是将整个表细分为许多小分区。 每个分区都有自己的锁。 因此，假设不同线程在表的不同分区上进行写入，则它们从不同线程对该映射的写入操作不会竞争，并且可以使用自己的锁。

该实现还引入了提交写操作的想法，以减少读操作的等待时间。 这将略微更改读取操作的语义，因为它将返回已完成的最新写入操作的结果。 这意味着在执行read方法之前和之后，条目的数量可能不一样，就像使用同步方法时一样，但是对于并发应用程序，这并不总是很重要。 ConcurrentHashMap的迭代器实现也是如此。

为了更好地了解Hashtable性能，同步的HashMap和ConcurrentHashMap的性能，让我们实现一个简单的性能测试。 以下代码启动了几个线程，并允许每个线程从映射中的一个随机位置检索一个值，然后在另一个随机位置更新一个值：

public class MapComparison implements Runnable {private static Map<Integer, String> map;private Random random = new Random(System.currentTimeMillis());public static void main(String[] args) throws InterruptedException {runPerfTest(new Hashtable<Integer, String>());runPerfTest(Collections.synchronizedMap(new HashMap<Integer,String>()));runPerfTest(new ConcurrentHashMap<Integer, String>());runPerfTest(new ConcurrentSkipListMap<Integer, String>());}private static void runPerfTest(Map<Integer, String> map) throws InterruptedException {MapComparison.map = map;fillMap(map);ExecutorService executorService = Executors.newFixedThreadPool(10);long startMillis = System.currentTimeMillis();for (int i = 0; i < 10; i++) {executorService.execute(new MapComparison());}executorService.shutdown();executorService.awaitTermination(1, TimeUnit.MINUTES);System.out.println(map.getClass().getSimpleName() + " took " + (System.currentTimeMillis() - startMillis) + " ms");}private static void fillMap(Map<Integer, String> map) {for (int i = 0; i < 100; i++) {map.put(i, String.valueOf(i));}}public void run() {for (int i = 0; i < 100000; i++) {int randomInt = random.nextInt(100);map.get(randomInt);randomInt = random.nextInt(100);map.put(randomInt, String.valueOf(randomInt));}}
}

该程序的输出如下：

Hashtable took 436 ms
SynchronizedMap took 433 ms
ConcurrentHashMap took 75 ms
ConcurrentSkipListMap took 89 ms

正如我们所期望的， Hashtable和同步的HashMap实现远远落后于并HashMap实现。 本示例还介绍了HashMap的跳过列表实现，其中一个存储桶中的链接项形成一个跳过列表，这意味着对列表进行了排序，并且列表中链接项的级别不同。 最高级别的指针直接指向列表中间的某个项目。 如果该项目已经大于当前项目，则迭代器必须采用下一个较低的链接级别，以跳过比最高级别更少的元素。 跳过列表的详细说明可以在此处找到。 关于跳过列表的有趣之处在于，即使所有项目都存储在同一存储桶中，所有读取访问也要花费log（n）时间。

原子变量

当多个线程共享一个变量时，我们需要同步对该变量的访问。 原因是这样的事实，即使像i ++这样的简单指令也不是原子的。 它基本上由以下字节码指令组成：

iload_1
iinc 1, 1
istore_1

在不了解Java字节码的情况下，人们看到了局部变量1的当前值被压入操作数堆栈，它以常数1递增，然后从堆栈中弹出并存储在局部变量号1中。 。 这意味着我们需要三个原子操作才能将局部变量加1。 在多线程环境中，这还意味着调度程序可以停止在这些指令中的每条指令之间执行当前线程，并启动一个新线程，然后该新线程又可以在同一变量上工作。

为了应对这种情况，您当然可以同步对此特定变量的访问：

synchronized(i) {i++;
}

但这也意味着当前线程必须获取i的锁，这需要在JVM中进行一些内部同步和计算。 这种方法也称为悲观锁定，因为我们认为另一个线程当前持有我们想要获取的锁定的可能性很高。 另一种称为乐观锁定的方法假定没有太多线程争用资源，因此我们只是尝试更新该值并查看是否起作用。 此方法的一种实现是比较交换（CAS）方法。 此操作在许多现代CPU上实现为原子操作。 它将给定存储位置的内容与给定值（“期望值”）进行比较，如果当前值等于期望值，则将其更新为新值。 用伪代码看起来像：

int currentValue = getValueAtMemoryPosition(pos);
if(currentValue == expectedValue) {setValueAtMemoryPosition(pos, newValue);
}

CAS操作将上述代码实现为一个原子操作。 因此，它可以用来查看某个变量的值是否仍为当前线程持有的值，并在这种情况下将其更新为递增的值。 由于CAS操作的使用需要硬件支持，因此JDK提供了特殊的类来支持这些操作。 它们都位于包java.util.concurrent.atomic中。

这些类的代表是java.util.concurrent.atomic.AtomicInteger 。 上面讨论的CAS操作是通过该方法实现的

boolean compareAndSet(int expect, int update)

布尔值返回值指示更新操作是否成功。 基于此功能，可以实现进一步的操作，例如原子增量操作（此处取自Oracle的JDK实现）：

public final int getAndIncrement() {for (;;) {int current = get();int next = current + 1;if (compareAndSet(current, next))return current;}}

现在我们可以通过不同的线程递增整数变量，而无需使用悲观锁：

public class AtomicIntegerExample implements Runnable {private static final AtomicInteger atomicInteger = new AtomicInteger();public static void main(String[] args) {ExecutorService executorService = Executors.newFixedThreadPool(5);for (int i = 0; i < 5; i++) {executorService.execute(new AtomicIntegerExample());}executorService.shutdown();}public void run() {for (int i = 0; i < 10; i++) {int newValue = atomicInteger.getAndIncrement();if (newValue == 42) {System.out.println("[" + Thread.currentThread().getName() + "]: " + newValue);}}}
}

上面的代码启动了五个线程，并让每个线程递增AtomicInteger变量。 得到答案的幸运线42将其打印到控制台。 重复执行此示例代码时，输​​出将仅由一个线程创建。

在AtomicInteger旁边，JDK还提供了用于对长值，整数和长数组以及引用进行原子操作的类。

信号

信号量用于控制对共享资源的访问。 与简单的同步块相反，信号量具有一个内部计数器，该内部计数器在线程每次获取锁时增加，而在线程释放其之前获得的锁时减少。 递增和递减操作当然是同步的，因此可以使用信号量来控制同时通过关键部分的线程数。 线程的两个基本操作是：

void acquire();
void release();

构造函数采用并发锁定公平性参数的数量。 fairness参数决定是否在等待线程列表的开头或结尾设置尝试获取锁的新线程。 将新线程放在线程末尾可确保所有线程在一段时间后将获得锁，因此不会出现线程饥饿的情况。

Semaphore(int permits, boolean fair)

为了说明所描述的行为，让我们建立一个具有五个线程的简单线程池，但通过一个信号量进行控制，在每个时间点运行的信号不超过三个：

public class SemaphoreExample implements Runnable {private static final Semaphore semaphore = new Semaphore(3, true);private static final AtomicInteger counter = new AtomicInteger();private static final long endMillis = System.currentTimeMillis() + 10000;public static void main(String[] args) {ExecutorService executorService = Executors.newFixedThreadPool(5);for (int i = 0; i < 5; i++) {executorService.execute(new SemaphoreExample());}executorService.shutdown();}public void run() {while(System.currentTimeMillis() < endMillis) {try {semaphore.acquire();} catch (InterruptedException e) {System.out.println("["+Thread.currentThread().getName()+"] Interrupted in acquire().");}int counterValue = counter.incrementAndGet();System.out.println("["+Thread.currentThread().getName()+"] semaphore acquired: "+counterValue);if(counterValue > 3) {throw new IllegalStateException("More than three threads acquired the lock.");}counter.decrementAndGet();semaphore.release();}}
}

通过将3作为并发许可的数量来构造信号量。 当尝试获取锁时，被阻止的线程可能会遇到必须捕获的InterruptedException 。 或者，也可以调用实用程序方法acquireUninterruptibly()来绕过try-catch构造。

为确保关键部分中的并发线程不超过三个，我们使用AtomicInteger ，每次进程进入该部分时该AtomicInteger都会递增，而在离开该部分之前会递减。 当计数器的值大于4时，将引发IllegalStateException 。 最后，我们release()信号量，然后让另一个等待线程进入临界区。

CountDownLatch

CountDownLatch类是另一个有助于从JDK进行线程同步的类。 类似于Semaphore类，它提供了一个计数器，但是CountDownLatch的计数器只能减少到零为止。 一旦计数器达到零，所有等待CountDownLatch线程都可以继续。 当池中的所有线程必须在某个点进行同步才能继续进行时，通常需要这种功能。 一个简单的示例是一个应用程序，该应用程序必须先从不同来源收集数据，然后才能将新数据集存储到数据库中。

以下代码演示了五个线程如何在随机时间内睡眠。 唤醒的每个线程都会对闩锁进行递减计数，然后等待闩锁变为零。 最后，所有线程输出它们已完成的输出。

public class CountDownLatchExample implements Runnable {private static final int NUMBER_OF_THREADS = 5;private static final CountDownLatch latch = new CountDownLatch(NUMBER_OF_THREADS);private static Random random = new Random(System.currentTimeMillis());public static void main(String[] args) {ExecutorService executorService = Executors.newFixedThreadPool(NUMBER_OF_THREADS);for (int i = 0; i < NUMBER_OF_THREADS; i++) {executorService.execute(new CountDownLatchExample());}executorService.shutdown();}public void run() {try {int randomSleepTime = random.nextInt(20000);System.out.println("[" + Thread.currentThread().getName() + "] Sleeping for " + randomSleepTime);Thread.sleep(randomSleepTime);latch.countDown();System.out.println("[" + Thread.currentThread().getName() + "] Waiting for latch.");latch.await();System.out.println("[" + Thread.currentThread().getName() + "] Finished.");} catch (InterruptedException e) {e.printStackTrace();}}
}

运行此示例时，您将看到输出“ Waiting for闩锁”。 在不同的时间点出现，但“完成”。 每个线程的消息立即一个接一个地打印。

循环屏障

与CountDownLatch ， CyclicBarrier类实现了一个计数器，该计数器在递减为零后可以重置。 所有线程必须调用其方法await()直到内部计数器设置为零为止。 等待的线程然后被唤醒并可以继续。 然后在内部将计数器重置为其原始值，并且整个过程可以再次开始：

public class CyclicBarrierExample implements Runnable {private static final int NUMBER_OF_THREADS = 5;private static AtomicInteger counter = new AtomicInteger();private static Random random = new Random(System.currentTimeMillis());private static final CyclicBarrier barrier = new CyclicBarrier(5, new Runnable() {public void run() {counter.incrementAndGet();}});public static void main(String[] args) {ExecutorService executorService = Executors.newFixedThreadPool(NUMBER_OF_THREADS);for (int i = 0; i < NUMBER_OF_THREADS; i++) {executorService.execute(new CyclicBarrierExample());}executorService.shutdown();}public void run() {try {while(counter.get() < 3) {int randomSleepTime = random.nextInt(10000);System.out.println("[" + Thread.currentThread().getName() + "] Sleeping for " + randomSleepTime);Thread.sleep(randomSleepTime);System.out.println("[" + Thread.currentThread().getName() + "] Waiting for barrier.");barrier.await();System.out.println("[" + Thread.currentThread().getName() + "] Finished.");}} catch (Exception e) {e.printStackTrace();}}
}

上面的示例与CountDownLatch非常相似，但是与前面的示例相反，我向run()方法添加了while循环。 这种run()实现使每个线程都能继续进行sleeping和await()过程，直到计数器为三。 还要注意提供给CyclicBarrier的构造函数的匿名Runnable（）实现。 每当障碍被触发时，其run()方法都会执行。 在这里，我们增加了并发线程检查的计数器。

3.下载源代码

您可以下载本课程的源代码： concurrency-4.zip

翻译自: https://www.javacodegeeks.com/2015/09/the-java-util-concurrent-package.html