目录
一、java.util.concurrent - Java并发工具包
六、具有优先级的阻塞队列PriorityBlockingQueue
BlockingDeque 继承自BlockingQueue
java.util.concurrent.ConcurrentMap
Java 5 添加了一个新的包到 Java 平台，java.util.concurrent 包。这个包包含有一系列能够让 Java 的并发编程变得更加简单轻松的类。在这个包被添加以前，你需要自己去动手实现自己的相关工具类。
java.util.concurrent 包里的 BlockingQueue 接口表示一个线程安全放入和提取实例的队列。本小节我将给你演示如何使用这个 BlockingQueue。
BlockingQueue 通常用于一个线程生产对象，而另外一个线程消费这些对象的场景。如下图：
一个线程往里边放，另外一个线程从里边取的一个 BlockingQueue。
一个线程将会持续生产新对象并将其插入到队列之中，直到队列达到它所能容纳的临界点。也就是说，它是有限的。
如果该阻塞队列到达了其临界点，负责生产的线程将会在往里边插入新对象时发生阻塞。它会一直处于阻塞之中，直到负责消费的线程从队列中拿走一个对象。
负责消费的线程将会一直从该阻塞队列中拿出对象。如果消费线程尝试去从一个空的队列中提取对象的话，这个消费线程将会发生阻塞，直到一个生产线程把一个对象丢进队列。
BlockingQueue 具有 4 组不同的方法用于插入、移除以及对队列中的元素进行检查。如果请求的操作不能得到立即执行的话，每个方法的表现也不同。这些方法如下：
| 操作 | 抛异常 | 特定值 | 阻塞 | 超时 |
|---|---|---|---|---|
| 插入 | add(o) | offer(o) | put(o) | offer(o, timeout, timeunit) |
| 移除 | remove(o) | poll(o) | take(o) | poll(timeout, timeunit) |
| 检查 | element(o) | peek(o) | 不可用 | 不可用 |
四组不同的行为方式解释：
无法向一个 BlockingQueue 中插入 null。如果你试图插入 null，BlockingQueue 将会抛出一个 NullPointerException。
可以访问到 BlockingQueue 中的所有元素，而不仅仅是开始和结束的元素。比如说，你将一个对象放入队列之中以等待处理，但你的应用想要将其取消掉。那么你可以调用诸如 remove(o) 方法来将队列之中的特定对象进行移除。但是这么干效率并不高(译者注：基于队列的数据结构，获取除开始或结束位置的其他对象的效率不会太高)，因此你尽量不要用这一类的方法，除非你确实不得不那么做。
BlockingQueue 是个接口，你需要使用它的实现之一来使用 BlockingQueue。java.util.concurrent 具有以下 BlockingQueue 接口的实现(Java 6)：
这里是一个 Java 中使用 BlockingQueue 的示例。本示例使用的是 BlockingQueue 接口的 ArrayBlockingQueue 实现。
首先，BlockingQueueExample 类分别在两个独立的线程中启动了一个 Producer 和 一个 Consumer。Producer 向一个共享的 BlockingQueue 中注入字符串，而 Consumer 则会从中把它们拿出来。
public class BlockingQueueExample {
public static void main(String[] args) throws Exception {
BlockingQueue queue = new ArrayBlockingQueue(1024);
Producer producer = new Producer(queue);
Consumer consumer = new Consumer(queue);
new Thread(producer).start();
new Thread(consumer).start();
Thread.sleep(4000);
}
} 以下是 Producer 类。注意它在每次 put() 调用时是如何休眠一秒钟的。这将导致 Consumer 在等待队列中对象的时候发生阻塞。
public class Producer implements Runnable{
protected BlockingQueue queue = null;
public Producer(BlockingQueue queue) {
this.queue = queue;
}
public void run() {
try {
queue.put("1");
Thread.sleep(1000);
queue.put("2");
Thread.sleep(1000);
queue.put("3");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
以下是 Consumer 类。它只是把对象从队列中抽取出来，然后将它们打印到 System.out
public class Consumer implements Runnable{
protected BlockingQueue queue = null;
public Consumer(BlockingQueue queue) {
this.queue = queue;
}
public void run() {
try {
System.out.println(queue.take());
System.out.println(queue.take());
System.out.println(queue.take());
} catch (InterruptedException e) {
e.printStackTrace();
}
}
} ArrayBlockingQueue 类实现了 BlockingQueue 接口。
ArrayBlockingQueue 是一个有界的阻塞队列，其内部实现是将对象放到一个数组里。有界也就意味着，它不能够存储无限多数量的元素。它有一个同一时间能够存储元素数量的上限。你可以在对其初始化的时候设定这个上限，但之后就无法对这个上限进行修改了(译者注：因为它是基于数组实现的，也就具有数组的特性：一旦初始化，大小就无法修改)。
ArrayBlockingQueue 内部以 FIFO(先进先出)的顺序对元素进行存储。队列中的头元素在所有元素之中是放入时间最久的那个，而尾元素则是最短的那个。
以下是在使用 ArrayBlockingQueue 的时候对其初始化的一个示例：
BlockingQueue queue = new ArrayBlockingQueue(1024);
queue.put("1");
Object object = queue.take(); 以下是使用了 Java 泛型的一个 BlockingQueue 示例。注意其中是如何对 String 元素放入和提取的：
BlockingQueue<String> queue = new ArrayBlockingQueue<String>(1024);
queue.put("1");
String string = queue.take(); 总结：延时任务队列的原理与实现_小魏的博客的博客-CSDN博客_延时任务队列
DelayQueue 实现了 BlockingQueue 接口。DelayQueue 对元素进行持有直到一个特定的延迟到期。
注入其中的元素必须实现java.util.concurrent.Delayed接口，该接口定义：
public interface Delayed extends Comparable<Delayed> {
public long getDelay(TimeUnit timeUnit);
} DelayQueue 将会在每个元素的 getDelay() 方法返回的值的时间段之后才释放掉该元素。如果返回的是 0 或者负值，延迟将被认为过期，该元素将会在 DelayQueue 的下一次 take 被调用的时候被释放掉。
传递给 getDelay 方法的 getDelay 实例是一个枚举类型，它表明了将要延迟的时间段。TimeUnit 枚举将会取以下值：
正如你所看到的，Delayed 接口也继承了 java.lang.Comparable 接口，这也就意味着 Delayed 对象之间可以进行对比。这个可能在对 DelayQueue 队列中的元素进行排序时有用，因此它们可以根据过期时间进行有序释放。
以下是使用 DelayQueue 的例子：
public class DelayQueueExample {
public static void main(String[] args) {
DelayQueue queue = new DelayQueue();
Delayed element1 = new DelayedElement();
queue.put(element1);
Delayed element2 = queue.take();
}
} DelayedElement 是我所创建的一个 DelayedElement 接口的实现类，它不在 Java.util.concurrent 包里。你需要自行创建你自己的 Delayed 接口的实现以使用 DelayQueue 类。
LinkedBlockingQueue 类实现了 BlockingQueue 接口。
LinkedBlockingQueue 内部以一个链式结构(链接节点)对其元素进行存储。如果需要的话，这一链式结构可以选择一个上限。如果没有定义上限，将使用 Integer.MAX_VALUE 作为上限。
LinkedBlockingQueue 内部以 FIFO(先进先出)的顺序对元素进行存储。队列中的头元素在所有元素之中是放入时间最久的那个，而尾元素则是最短的那个。
以下是 LinkedBlockingQueue 的初始化和使用示例代码：
BlockingQueue<String> unbounded = new LinkedBlockingQueue<String>();
BlockingQueue<String> bounded = new LinkedBlockingQueue<String>(1024);
bounded.put("Value");
String value = bounded.take(); PriorityBlockingQueue 类实现了 BlockingQueue 接口。
PriorityBlockingQueue 是一个无界的并发队列。它使用了和类 java.util.PriorityQueue 一样的排序规则。你无法向这个队列中插入 null 值。
所有插入到 PriorityBlockingQueue 的元素必须实现 java.lang.Comparable 接口。因此该队列中元素的排序就取决于你自己的 Comparable 实现。
注意 PriorityBlockingQueue 对于具有相等优先级(compare() == 0)的元素并不强制任何特定行为。
同时注意，如果你从一个 PriorityBlockingQueue 获得一个 Iterator 的话，该 Iterator 并不能保证它对元素的遍历是以优先级为序的。
以下是使用 PriorityBlockingQueue 的示例：
BlockingQueue queue = new PriorityBlockingQueue();
//String implements java.lang.Comparable
queue.put("Value");
String value = queue.take(); SynchronousQueue 类实现了 BlockingQueue 接口。
SynchronousQueue 是一个特殊的队列，它的内部同时只能够容纳单个元素。如果该队列已有一元素的话，试图向队列中插入一个新元素的线程将会阻塞，直到另一个线程将该元素从队列中抽走。同样，如果该队列为空，试图向队列中抽取一个元素的线程将会阻塞，直到另一个线程向队列中插入了一条新的元素。
据此，把这个类称作一个队列显然是夸大其词了。它更多像是一个汇合点。
在线程既是一个队列的生产者又是这个队列的消费者的时候可以使用到 BlockingDeque。如果生产者线程需要在队列的两端都可以插入数据，消费者线程需要在队列的两端都可以移除数据，这个时候也可以使用 BlockingDeque。BlockingDeque 图解：
一个 BlockingDeque - 线程在双端队列的两端都可以插入和提取元素。
一个线程生产元素，并把它们插入到队列的任意一端。如果双端队列已满，插入线程将被阻塞，直到一个移除线程从该队列中移出了一个元素。如果双端队列为空，移除线程将被阻塞，直到一个插入线程向该队列插入了一个新元素。
BlockingDeque 具有 4 组不同的方法用于插入、移除以及对双端队列中的元素进行检查。如果请求的操作不能得到立即执行的话，每个方法的表现也不同。这些方法如下：
| 操作 | 抛异常 | 特定值 | 阻塞 | 超时 |
|---|---|---|---|---|
| 插入 | addFirst(o) | offerFirst(o) | putFirst(o) | offerFirst(o, timeout, timeunit) |
| 移除 | removeFirst(o) | pollFirst(o) | takeFirst(o) | pollFirst(timeout, timeunit) |
| 检查 | getFirst(o) | peekFirst(o) | 无 | 无 |
| 操作 | 抛异常 | 特定值 | 阻塞 | 超时 |
|---|---|---|---|---|
| 插入 | addLast(o) | offerLast(o) | putLast(o) | offerLast(o, timeout, timeunit) |
| 移除 | removeLast(o) | pollLast(o) | takeLast(o) | pollLast(timeout, timeunit) |
| 检查 | getLast(o) | peekLast(o) | 无 | 无 |
四组不同的行为方式解释：
BlockingDeque 接口继承自 BlockingQueue 接口。这就意味着你可以像使用一个 BlockingQueue 那样使用BlockingDeque。如果你这么干的话，各种插入方法将会把新元素添加到双端队列的尾端，而移除方法将会把双端队列的首端的元素移除。正如 BlockingQueue 接口的插入和移除方法一样。
以下是 BlockingDeque 对 BlockingQueue 接口的方法的具体内部实现：
| BlockingQueue | BlockingDeque |
|---|---|
| add() | addLast() |
| offer() | offerLast() |
| put() | putLast() |
| offer(e, time, unit) | offerLast(e, time, unit) |
| remove() | removeFirst() |
| poll() | pollFirst() |
| take() | takeFirst() |
| poll(time, unit) | pollLast(time, unit) |
| element() | getFirst() |
| peek() | peekFirst() |
既然 BlockingDeque 是一个接口，那么你想要使用它的话就得使用它的众多的实现类的其中一个。java.util.concurrent 包提供了以下 BlockingDeque 接口的实现类：LinkedBlockingDeque
以下是如何使用 BlockingDeque 方法的一个简短代码示例：
BlockingDeque<String> deque = new LinkedBlockingDeque<String>();
deque.addFirst("1");
deque.addLast("2");
String two = deque.takeLast();
String one = deque.takeFirst(); LinkedBlockingDeque 类实现了 BlockingDeque 接口。
deque(双端队列) 是 “Double Ended Queue” 的缩写。因此，双端队列是一个你可以从任意一端插入或者抽取元素的队列。
LinkedBlockingDeque 是一个双端队列，在它为空的时候，一个试图从中抽取数据的线程将会阻塞，无论该线程是试图从哪一端抽取数据。
以下是 LinkedBlockingDeque 实例化以及使用的示例：
BlockingDeque<String> deque = new LinkedBlockingDeque<String>();
deque.addFirst("1");
deque.addLast("2");
String two = deque.takeLast();
String one = deque.takeFirst(); java.util.concurrent.ConcurrentMap 接口表示了一个能够对别人的访问(插入和提取)进行并发处理的 java.util.Map。
ConcurrentMap 除了从其父接口 java.util.Map 继承来的方法之外还有一些额外的原子性方法。
既然 ConcurrentMap 是个接口，你想要使用它的话就得使用它的实现类之一。java.util.concurrent 包具备 ConcurrentMap 接口的以下实现类：ConcurrentHashMap
ConcurrentHashMap 和 java.util.HashTable 类很相似，但 ConcurrentHashMap 能够提供比 HashTable 更好的并发性能。在你从中读取对象的时候 ConcurrentHashMap 并不会把整个 Map 锁住。此外，在你向其中写入对象的时候，ConcurrentHashMap 也不会锁住整个 Map。它的内部只是把 Map 中正在被写入的部分进行锁定。
另外一个不同点是，在被遍历的时候，即使是 ConcurrentHashMap 被改动，它也不会抛ConcurrentModificationException。尽管 Iterator 的设计不是为多个线程的同时使用。
以下是如何使用 ConcurrentMap 接口的一个例子。本示例使用了 ConcurrentHashMap 实现类：
ConcurrentMap concurrentMap = new ConcurrentHashMap();
concurrentMap.put("key", "value");
Object value = concurrentMap.get("key"); java.util.concurrent.ConcurrentNavigableMap 是一个支持并发访问的 java.util.NavigableMap，它还能让它的子 map 具备并发访问的能力。所谓的 “子 map” 指的是诸如 headMap()，subMap()，tailMap() 之类的方法返回的 map。
NavigableMap 中的方法不再赘述，本小节我们来看一下 ConcurrentNavigableMap 添加的方法。
headMap(T toKey) 方法返回一个包含了小于给定 toKey 的 key 的子 map。
如果你对原始 map 里的元素做了改动，这些改动将影响到子 map 中的元素(译者注：map 集合持有的其实只是对象的引用)。
以下示例演示了对 headMap() 方法的使用：
ConcurrentNavigableMap map = new ConcurrentSkipListMap();
map.put("1", "one");
map.put("2", "two");
map.put("3", "three");
ConcurrentNavigableMap headMap = map.headMap("2"); headMap 将指向一个只含有键 “1” 的 ConcurrentNavigableMap，因为只有这一个键小于 “2”。关于这个方法及其重载版本具体是怎么工作的细节请参考 Java 文档。
tailMap(T fromKey) 方法返回一个包含了不小于给定 fromKey 的 key 的子 map。
如果你对原始 map 里的元素做了改动，这些改动将影响到子 map 中的元素(译者注：map 集合持有的其实只是对象的引用)。
以下示例演示了对 tailMap() 方法的使用：
ConcurrentNavigableMap map = new ConcurrentSkipListMap();
map.put("1", "one");
map.put("2", "two");
map.put("3", "three");
ConcurrentNavigableMap tailMap = map.tailMap("2"); tailMap 将拥有键 “2” 和 “3”，因为它们不小于给定键 “2”。关于这个方法及其重载版本具体是怎么工作的细节请参考 Java 文档。
subMap() 方法返回原始 map 中，键介于 from(包含) 和 to (不包含) 之间的子 map。示例如下：
ConcurrentNavigableMap map = new ConcurrentSkipListMap();
map.put("1", "one");
map.put("2", "two");
map.put("3", "three");
ConcurrentNavigableMap subMap = map.subMap("2", "3"); 返回的 submap 只包含键 “2”，因为只有它满足不小于 “2”，比 “3” 小。
ConcurrentNavigableMap 接口还有其他一些方法可供使用，比如：
关于这些方法更多信息参考官方 Java 文档。
总结：Java并发编程之CountDownLatch与CyclicBarrier_小魏的博客的博客-CSDN博客
java.util.concurrent.CountDownLatch 是一个并发构造，它允许一个或多个线程等待一系列指定操作的完成。
CountDownLatch 以一个给定的数量初始化。countDown() 每被调用一次，这一数量就减一。通过调用 await() 方法之一，线程可以阻塞等待这一数量到达零。
以下是一个简单示例。Decrementer 三次调用 countDown() 之后，等待中的 Waiter 才会从 await() 调用中释放出来。
public class Test {
public static void main(String[] args) throws Exception{
// 定义CountDownLatch ， 计数器数量为“子任务的个数”
final CountDownLatch latch = new CountDownLatch(5);
List<StringBuilder> stringBuilders = new ArrayList<>();
stringBuilders.add(new StringBuilder("111"));
stringBuilders.add(new StringBuilder("222"));
stringBuilders.add(new StringBuilder("333"));
stringBuilders.add(new StringBuilder("444"));
stringBuilders.add(new StringBuilder("555"));
ThreadPoolManager threadPoolManager = ThreadPoolManager.newInstance();
for (StringBuilder stringBuilder : stringBuilders) {
threadPoolManager.addExecuteTask(() -> {
try {
handle(stringBuilder, latch);
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
Console.log("{}:等待任务执行完毕", new Date());
latch.await();
Console.log("{}:全部任务已经执行完毕", new Date());
Console.log("stringBuilders:{}", JSONObject.toJSONString(stringBuilders));
}
public static void handle(StringBuilder stringBuilder, CountDownLatch latch) throws InterruptedException {
String name = Thread.currentThread().getName();
stringBuilder.append("#666");
Thread.sleep(3000);
Console.log("线程：{} 执行完毕！", name);
latch.countDown();
}
}java.util.concurrent.CyclicBarrier 类是一种同步机制，它能够对处理一些算法的线程实现同步。换句话讲，它就是一个所有线程必须等待的一个栅栏，直到所有线程都到达这里，然后所有线程才可以继续做其他事情
java.util.concurrent.Exchanger 类表示一种两个线程可以进行互相交换对象的会和点。这种机制图示如下：
两个线程通过一个 Exchanger 交换对象。
交换对象的动作由 Exchanger 的两个 exchange() 方法的其中一个完成。以下是一个示例：
class Person{
String name;
String age;
public Person(String name, String age) {
super();
this.name = name;
this.age = age;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public String getAge() {
return age;
}
public void setAge(String age) {
this.age = age;
}
}
class ExchangerRunnable implements Runnable{
Exchanger<Person> exchanger = null;
Person object = null;
public ExchangerRunnable(Exchanger<Person> exchanger, Person object) {
this.exchanger = exchanger;
this.object = object;
}
public void run() {
try {
Person previous = this.object;
this.object = this.exchanger.exchange(this.object);
System.out.println(
Thread.currentThread().getName() +
" 的老师由 " + previous.getName() + " 变为 " + this.object.getName()
);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
public class ExchangerTest {
public static void main(String[] args) {
Exchanger<Person> exchanger = new Exchanger<Person>();
Person teacherZhang = new Person("张老师", "50");
Person teacherLiu = new Person("刘老师", "37");
//起初老师为张老师
ExchangerRunnable exchangerRunnable1 = new ExchangerRunnable(exchanger, teacherZhang);
//起初老师为刘老师
ExchangerRunnable exchangerRunnable2 = new ExchangerRunnable(exchanger, teacherLiu);
new Thread(exchangerRunnable1, "weiwei").start();
new Thread(exchangerRunnable2, "zhangyu").start();
}
}以上程序输出：
weiwei 的老师由 张老师 变为 刘老师
zhangyu 的老师由 刘老师 变为 张老师java.util.concurrent.Semaphore 类是一个计数信号量。
Semaphore可以控同时访问的线程个数，通过 acquire() 获取一个许可，如果没有就等待，而 release() 释放一个许可。
案例：
假若一个工厂有5台机器，但是有8个工人，一台机器同时只能被一个工人使用，只有使用完了，其他工人才
能继续使用。那么我们就可以通过Semaphore来实现：
public class SemaphoreTest {
public static void main(String[] args) {
int N = 8; //工人数
Semaphore semaphore = new Semaphore(5); //机器数目
for(int i=0;i<N;i++)
new Worker(i,semaphore).start();
}
}
class Worker extends Thread{
private int num;
private Semaphore semaphore;
public Worker(int num,Semaphore semaphore){
this.num = num;
this.semaphore = semaphore;
}
@Override
public void run() {
try {
semaphore.acquire();
System.out.println("工人"+this.num+"占用一个机器在生产...");
Thread.sleep(2000);
System.out.println("工人"+this.num+"释放出机器");
semaphore.release();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}输出：
工人0占用一个机器在生产...
工人2占用一个机器在生产...
工人1占用一个机器在生产...
工人3占用一个机器在生产...
工人4占用一个机器在生产...
工人4释放出机器
工人3释放出机器
工人5占用一个机器在生产...
工人6占用一个机器在生产...
工人0释放出机器
工人2释放出机器
工人7占用一个机器在生产...
工人1释放出机器
工人5释放出机器
工人6释放出机器
工人7释放出机器java.util.concurrent.ExecutorService是Java中对线程池定义的一个接口。
以下是一个简单的 ExecutorService 例子：
ExecutorService executorService = Executors.newFixedThreadPool(10);
executorService.execute(new Runnable() {
public void run() {
System.out.println("Asynchronous task");
}
});
executorService.shutdown(); 其它例子：
ExecutorService executorService1 = Executors.newSingleThreadExecutor();
ExecutorService executorService2 = Executors.newFixedThreadPool(10);
ExecutorService executorService3 = Executors.newScheduledThreadPool(10); 
既然 ExecutorService 是个接口，如果你想用它的话就得去使用它的实现类之一。java.util.concurrent 包提供了 ExecutorService 接口的以下实现类：
接下来我们挨个看一下这些方法。
execute(Runnable)
execute(Runnable) 方法要求一个Runnable对象，然后对它进行异步执行。以下是使用 ExecutorService 执行一个 Runnable 的示例：
注：execute方法是在定级接口Executor中定义的，ExecutorService继承了Executor。
ExecutorService executorService = Executors.newSingleThreadExecutor();
executorService.execute(new Runnable() {
public void run() {
System.out.println("Asynchronous task");
}
});
executorService.shutdown(); 没有办法得知被执行的 Runnable 的执行结果。如果有需要的话你得使用一个 Callable(以下将做介绍)。
6、submit(Runnable)
submit(Runnable) 方法也要求一个 Runnable 实现类，但它返回一个 Future 对象。这个 Future 对象可以用来检查 Runnable 是否已经执行完毕。
以下是 ExecutorService submit() 示例：
Future future = executorService.submit(new Runnable() {
public void run() {
System.out.println("Asynchronous task");
}
});
future.get(); //returns null if the task has finished correctly. 7、submit(Callable)
submit(Callable) 方法类似于 submit(Runnable) 方法，除了它所要求的参数类型之外。Callable 实例除了它的 call() 方法能够返回一个结果之外和一个 Runnable 很相像。Runnable.run() 不能够返回一个结果。
Callable 的结果可以通过 submit(Callable) 方法返回的 Future 对象进行获取。以下是一个 ExecutorService Callable 示例：
Future future = executorService.submit(new Callable(){
public Object call() throws Exception {
System.out.println("Asynchronous Callable");
return "Callable Result";
}
});
System.out.println("future.get() = " + future.get()); 以上代码输出：
Asynchronous Callable
future.get() = Callable Result8、ExecutorService 关闭
如果是临时创建的ExecutorService，使用完 ExecutorService 之后应该将其关闭，释放资源。
使用 ExecutorService 类时，经常用到shutdown() 、shutdownNow()2个方法
9、invokeAny()
invokeAny(…)方法接收的是一个Callable的集合，执行这个方法不会返回Future，但是会返回所有Callable任务中其中一个任务的执行结果。这个方法也无法保证返回的是哪个任务的执行结果，反正是其中的某一个。
以下是示例代码：
ExecutorService executorService = Executors.newSingleThreadExecutor();
Set<Callable<String>> callables = new HashSet<Callable<String>>();
callables.add(new Callable<String>() {
public String call() throws Exception {
return "Task 1";
}
});
callables.add(new Callable<String>() {
public String call() throws Exception {
return "Task 2";
}
});
callables.add(new Callable<String>() {
public String call() throws Exception {
return "Task 3";
}
});
String result = executorService.invokeAny(callables);
System.out.println("result = " + result);
executorService.shutdown();10、invokeAll()
invokeAll(…)与 invokeAny(…)类似也是接收一个Callable集合，但是invokeAll(…)执行之后会返回一个Future的List，其中对应着每个Callable任务执行后的Future对象。
ThreadPoolExecutor 是 ExecutorService 接口的一个实现。ThreadPoolExecutor 使用其内部池中的线程执行给定任务(Callable 或者 Runnable)。
ThreadPoolExecutor包含的线程池能够包含不同数量的线程。池中线程的数量由以下变量决定：
当一个任务委托给线程池时，如果池中线程数量低于 corePoolSize，一个新的线程将被创建，即使池中可能尚有空闲线程。如果内部任务队列已满，而且有至少 corePoolSize 正在运行，但是运行线程的数量低于 maximumPoolSize，一个新的线程将被创建去执行该任务。
创建一个 ThreadPoolExecutor
int corePoolSize = 5;
int maxPoolSize = 10;
long keepAliveTime = 5000;
ExecutorService threadPoolExecutor =
new ThreadPoolExecutor(
corePoolSize,
maxPoolSize,
keepAliveTime,
TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()
); java.util.concurrent.ScheduledExecutorService 是一个 ExecutorService。
它能够将任务延后执行，或者间隔固定时间多次执行。 任务由一个工作者线程异步执行，而不是由提交任务给 ScheduledExecutorService 的那个线程执行。
以下是一个简单的 ScheduledExecutorService 示例：
ScheduledExecutorService scheduledExecutorService =
Executors.newScheduledThreadPool(5);
ScheduledFuture scheduledFuture =
scheduledExecutorService.schedule(new Callable() {
public Object call() throws Exception {
System.out.println("Executed!");
return "Called!";
}
}, 5, TimeUnit.SECONDS); 首先一个内置 5 个线程的 ScheduledExecutorService 被创建。之后一个 Callable 接口的匿名类示例被创建然后传递给 schedule() 方法。后边的俩参数定义了 Callable 将在 5 秒钟之后被执行。
2、ScheduledExecutorService 实现
既然 ScheduledExecutorService 是一个接口，你要用它的话就得使用 java.util.concurrent 包里对它的某个实现类。ScheduledExecutorService 具有以下实现类：ScheduledThreadPoolExecutor
如何创建一个 ScheduledExecutorService 取决于你采用的它的实现类。但是你也可以使用 Executors 工厂类来创建一个 ScheduledExecutorService 实例。比如：
ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(5); 一旦你创建了一个 ScheduledExecutorService，你可以通过调用它的以下方法：
下面我们就简单看一下这些方法。
schedule (Callable task, long delay, TimeUnit timeunit)这个方法计划指定的 Callable 在给定的延迟之后执行。
这个方法返回一个 ScheduledFuture，通过它你可以在它被执行之前对它进行取消，或者在它执行之后获取结果。
以下是一个示例：
ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(5);
ScheduledFuture scheduledFuture =
scheduledExecutorService.schedule(new Callable() {
public Object call() throws Exception {
System.out.println("Executed!");
return "Called!";
}
}, 5, TimeUnit.SECONDS);
System.out.println("result = " + scheduledFuture.get());
scheduledExecutorService.shutdown(); 示例输出结果：
Executed!
result = Called!
4、ScheduledExecutorService 关闭
可以使用从 ExecutorService 接口继承来的 shutdown() 或 shutdownNow() 方法将 ScheduledExecutorService 关闭。参见 ExecutorService 关闭部分以获取更多信息。
总结：Java中的锁_小魏的博客的博客-CSDN博客_java中的锁
java.util.concurrent.locks.Lock是一个类似于 synchronized 块的线程同步机制。但是 Lock 比 synchronized 块更加灵活、精细。
既然 Lock 是一个接口，在你的程序里需要使用它的实现类之一来使用它。以下是一个简单示例：
Lock lock = new ReentrantLock();
lock.lock();
//......
lock.unlock(); 首先创建了一个 Lock 对象。之后调用了它的 lock() 方法。这时候这个 lock 实例就被锁住啦。任何其他再过来调用 lock() 方法的线程将会被阻塞住，直到锁定 lock 实例的线程调用了 unlock() 方法。最后 unlock() 被调用了，lock 对象解锁了，其他线程可以对它进行锁定了。
Lock 和 synchronized对比：
一个 Lock 对象和一个 synchronized 代码块之间的主要不同点是：
Lock 的方法
Lock 接口具有以下主要方法：
lock() 将 Lock 实例锁定。如果该 Lock 实例已被锁定，调用 lock() 方法的线程将会阻塞，直到 Lock 实例解锁。
lockInterruptibly() 方法将会被调用线程锁定，除非该线程被打断。此外，如果一个线程在通过这个方法来锁定 Lock 对象时进入阻塞等待，而它被打断了的话，该线程将会退出这个方法调用。
tryLock() 方法试图立即锁定 Lock 实例。如果锁定成功，它将返回 true，如果 Lock 实例已被锁定该方法返回 false。这一方法永不阻塞。
tryLock(long timeout, TimeUnit timeUnit) 的工作类似于 tryLock() 方法，除了它在放弃锁定 Lock 之前等待一个给定的超时时间之外。
unlock() 方法对 Lock 实例解锁。一个 Lock 实现将只允许锁定了该对象的线程来调用此方法。其他(没有锁定该 Lock 对象的线程)线程对 unlock() 方法的调用将会抛一个未检查异常(RuntimeException)。
ReentrantReadWriteLock实现了ReadWriteLock接口。
ReadWriteLock接口包含两个方法：
public interface ReadWriteLock {
/**
* Returns the lock used for reading.
*/
Lock readLock();
/**
* Returns the lock used for writing.
*/
Lock writeLock();
}ReentrantReadWriteLock 是读写锁，和ReentrantLock会有所不同，对于读多写少的场景使用ReentrantReadWriteLock 性能会比ReentrantLock高出不少。在多线程读时互不影响，不像ReentrantLock即使是多线程读也需要每个线程获取锁。不过任何一个线程在写的时候就和ReentrantLock类似，其他线程无论读还是写都必须获取锁。需要注意的是同一个线程可以拥有 writeLock 与 readLock (但必须先获取 writeLock 再获取 readLock, 反过来进行获取会导致死锁)
案例：
private static final ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
public static void main(String[] args) {
Integer test = 2;
ExecutorService threadPool = Executors.newFixedThreadPool(10);
for (int i = 0; i < 3; i++) {
threadPool.execute(() -> {
try {
readWriteLock.writeLock().lockInterruptibly();
System.out.println("写入数据" + test);
// 假装耗时操作
TimeUnit.SECONDS.sleep(5);
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
readWriteLock.writeLock().unlock();
}
});
}
for (int i = 0; i < 20; i++) {
threadPool.execute(() -> {
try {
readWriteLock.readLock().lockInterruptibly();
System.out.println("读取数据" + test);
// 假装耗时操作
TimeUnit.SECONDS.sleep(5);
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
readWriteLock.readLock().unlock();
}
});
}
}AtomicBoolean 类拥有一些先进的原子性操作，比如 compareAndSet()。
修改AtomicBoolean 的值
你可以通过 getAndSet() 方法来交换一个 AtomicBoolean 实例的值。getAndSet() 方法将返回 AtomicBoolean 当前的值，并将为 AtomicBoolean 设置一个新值。示例如下：
AtomicBoolean atomicBoolean = new AtomicBoolean(true);
boolean oldValue = atomicBoolean.getAndSet(false); 以上代码执行后 oldValue 变量的值为 true，atomicBoolean 实例将持有 false 值。代码成功将 AtomicBoolean 当前值 ture 交换为 false。
比较并设置 AtomicBoolean 的值
compareAndSet() 方法允许你对 AtomicBoolean 的当前值与一个期望值进行比较，如果当前值等于期望值的话，将会对 AtomicBoolean 设定一个新值。compareAndSet() 方法是原子性的，因此在同一时间之内有单个线程执行它。因此 compareAndSet() 方法可被用于一些类似于锁的同步的简单实现。
以下是一个 compareAndSet() 示例：
AtomicBoolean atomicBoolean = new AtomicBoolean(true);
boolean expectedValue = true;
boolean newValue = false;
boolean wasNewValueSet = atomicBoolean.compareAndSet(expectedValue, newValue); 本示例对 AtomicBoolean 的当前值与 true 值进行比较，如果相等，将 AtomicBoolean 的值更新为 false。
AtomicInteger 类为我们提供了一个可以进行原子性读和写操作的 int 变量，它还包含一系列先进的原子性操作，比如 compareAndSet()。
AtomicInteger atomicInteger = new AtomicInteger(123);
int theValue = atomicInteger.get(); 设置 AtomicInteger 的值
你可以通过 set() 方法对 AtomicInteger 的值进行重新设置。以下是 AtomicInteger.set() 示例：
AtomicInteger atomicInteger = new AtomicInteger(123);
atomicInteger.set(234); 以上示例创建了一个初始值为 123 的 AtomicInteger，而在第二行将其值更新为 234。
比较并设置 AtomicInteger 的值
AtomicInteger 类也通过了一个原子性的 compareAndSet() 方法。这一方法将 AtomicInteger 实例的当前值与期望值进行比较，如果二者相等，为 AtomicInteger 实例设置一个新值。AtomicInteger.compareAndSet() 代码示例：
AtomicInteger atomicInteger = new AtomicInteger(123);
int expectedValue = 123;
int newValue = 234;
atomicInteger.compareAndSet(expectedValue, newValue); 本示例首先新建一个初始值为 123 的 AtomicInteger 实例。然后将 AtomicInteger 与期望值 123 进行比较，如果相等，将 AtomicInteger 的值更新为 234。
增加 AtomicInteger 值
AtomicInteger 类包含有一些方法，通过它们你可以增加 AtomicInteger 的值，并获取其值。这些方法如下：
减小 AtomicInteger 的值
AtomicInteger 类还提供了一些减小 AtomicInteger 的值的原子性方法。这些方法是：
decrementAndGet() 将 AtomicInteger 的值减一，并返回减一后的值。getAndDecrement() 也将 AtomicInteger 的值减一，但它返回的是减一之前的值。
AtomicLong 类为我们提供了一个可以进行原子性读和写操作的 long 变量，它还包含一系列先进的原子性操作，比如 compareAndSet()。
类似于AtomicInteger 类。
AtomicReference 提供了一个可以被原子性读和写的对象引用变量。原子性的意思是多个想要改变同一个 AtomicReference 的线程不会导致 AtomicReference 处于不一致的状态。AtomicReference 还有一个 compareAndSet() 方法，通过它你可以将当前引用于一个期望值(引用)进行比较，如果相等，在该 AtomicReference 对象内部设置一个新的引用。
创建一个 AtomicReference
创建 AtomicReference 如下：
AtomicReference atomicReference = new AtomicReference(); 如果你需要使用一个指定引用创建 AtomicReference，可以：
String initialReference = "the initially referenced string";
AtomicReference atomicReference = new AtomicReference(initialReference); 创建泛型 AtomicReference
你可以使用 Java 泛型来创建一个泛型 AtomicReference。示例：
AtomicReference<String> atomicStringReference = new AtomicReference<String>(); 你也可以为泛型 AtomicReference 设置一个初始值。示例：
String initialReference = "the initially referenced string";
AtomicReference<String> atomicStringReference = new AtomicReference<String>(initialReference); 获取 AtomicReference 引用
你可以通过 AtomicReference 的 get() 方法来获取保存在 AtomicReference 里的引用。如果你的 AtomicReference 是非泛型的，get() 方法将返回一个 Object 类型的引用。如果是泛型化的，get() 将返回你创建 AtomicReference 时声明的那个类型。
先来看一个非泛型的 AtomicReference get() 示例：
String initialReference = "the initially referenced string";
AtomicReference<String> atomicReference = new AtomicReference<String>(initialReference);
String reference = atomicReference.get();
System.out.println(reference);
//输出
//the initially referenced string注意如何对 get() 方法返回的引用强制转换为 String。
泛型化的 AtomicReference 示例：
AtomicReference<String> atomicReference = new AtomicReference<String>("first value referenced");
String reference = atomicReference.get(); 编译器知道了引用的类型，所以我们无需再对 get() 返回的引用进行强制转换了。
设置 AtomicReference 引用
你可以使用 get() 方法对 AtomicReference 里边保存的引用进行设置。如果你定义的是一个非泛型 AtomicReference，set() 将会以一个 Object 引用作为参数。如果是泛型化的 AtomicReference，set() 方法将只接受你定义给的类型。
AtomicReference set() 示例：
AtomicReference atomicReference = new AtomicReference();
atomicReference.set("New object referenced"); 这个看起来非泛型和泛型化的没啥区别。真正的区别在于编译器将对你能够设置给一个泛型化的 AtomicReference 参数类型进行限制。
比较并设置 AtomicReference 引用
AtomicReference 类具备了一个很有用的方法：compareAndSet()。compareAndSet() 可以将保存在 AtomicReference 里的引用于一个期望引用进行比较，如果两个引用是一样的(并非 equals() 的相等，而是 == 的一样)，将会给AtomicReference 实例设置一个新的引用。
如果 compareAndSet() 为 AtomicReference 设置了一个新的引用，compareAndSet() 将返回 true。否则compareAndSet() 返回 false。
AtomicReference compareAndSet() 示例：
String initialReference = "initial value referenced";
AtomicReference<String> atomicStringReference = new AtomicReference<String>(initialReference);
String newReference = "new value referenced";
boolean exchanged = atomicStringReference.compareAndSet(initialReference, newReference);
System.out.println("exchanged: " + exchanged);
exchanged = atomicStringReference.compareAndSet(initialReference, newReference);
System.out.println("exchanged: " + exchanged); 本示例创建了一个带有一个初始引用的泛型化的 AtomicReference。之后两次调用 comparesAndSet()来对存储值和期望值进行对比，如果二者一致，为 AtomicReference 设置一个新的引用。第一次比较，存储的引用(initialReference)和期望的引用(initialReference)一致，所以一个新的引用(newReference)被设置给 AtomicReference，compareAndSet() 方法返回 true。第二次比较时，存储的引用(newReference)和期望的引用(initialReference)不一致，因此新的引用没有被设置给 AtomicReference，compareAndSet() 方法返回 false。
参考：