简介
concurrent：并发
JUC：java.util.concurrent
用于并发控制编程
下面有两个包：atomic,locks
java.util.concurrent
它下面就有很多的类
collections部分:
BlockingQueue
阻塞队列
BlockingQueue为接口，如果要是它，需要使用实现他的子类；
BlockingQueue的子类包括'
ArrayBlockingQueue;
DelayQueue;
LinkedBlockingQueue;
SynchronousQueue;
PriorirtyBlockingQueue;
TransferQueue;
BlockingQueue 通常用于一个线程生产对象，而另外一个线程消费这些对象的场景。
一个线程将会持续生产新对象并将其插入到队列之中，直到队列达到它所能容纳的临界点。也就是说，它是有限的。如果该阻塞队列到达了其临界点，负责生产的线程将会在往里边插入新对象时发生阻塞。它会一直处于阻塞之中，直到负责消费的线程从队列中拿走一个对象。负责消费的线程将会一直从该阻塞队列中拿出对象。如果消费线程尝试去从一个空的队列中提取对象的话，这个消费线程将会处于阻塞之中，直到一个生产线程把一个对象丢进队列。
lockingQueue 具有 4 组不同的方法用于插入、移除以及对队列中的元素进行检查。如果请求的操作不能得到立即执行的话，每个方法的表现也不同。这些方法如下：
|
|
||||
|
|
抛异常 |
特定值 |
阻塞 |
超时 |
|
插入 |
add(o) |
offer(o) |
put(o) |
offer(o, timeout, timeunit) |
|
移除 |
remove(o) |
poll(o) |
take(o) |
poll(timeout, timeunit) |
|
检查 |
element(o) |
peek(o) |
|
|
四组不同的行为方式解释：
抛异常：如果试图的操作无法立即执行，抛一个异常。
特定值：如果试图的操作无法立即执行，返回一个特定的值(常常是 true / false)。
阻塞：如果试图的操作无法立即执行，该方法调用将会发生阻塞，直到能够执行。
超时：如果试图的操作无法立即执行，该方法调用将会发生阻塞，直到能够执行，但等待时间不会超过给定值。返回一个特定值以告知该操作是否成功(典型的是 true / false)。
无法向一个 BlockingQueue 中插入 null。如果你试图插入 null，BlockingQueue 将会抛出一个 NullPointerException。
可以访问到 BlockingQueue 中的所有元素，而不仅仅是开始和结束的元素。比如说，你将一个对象放入队列之中以等待处理，但你的应用想要将其取消掉。那么你可以调用诸如 remove(o) 方法来将队列之中的特定对象进行移除。但是这么干效率并不高(译者注：基于队列的数据结构，获取除开始或结束位置的其他对象的效率不会太高)，因此你尽量不要用这一类的方法，除非你确实不得不那么做。
BlockingQueue内部是拿数组存储，那么上限就是Integer.MAX_VALUE，也就是2147483647个元素
DelayQueue
特点： 无界、延迟、阻塞队列
BlockingQueue+PriorityQueue（堆排序）+Delayed
DelayQueue 实现了 BlockingQueue 接口。
DelayQueue 对元素进行持有直到一个特定的延迟到期。注入其中的元素必须实现 java.util.concurrent.Delayed 接口
DelayQueue 将会在每个元素的 getDelay() 方法返回的值的时间段之后才释放掉该元素。如果返回的是 0 或者负值，延迟将被认为过期，该元素将会在 DelayQueue 的下一次 take 被调用的时候被释放掉。
传递给 getDelay 方法的 getDelay 实例是一个枚举类型，它表明了将要延迟的时间段。TimeUnit 枚举将会取以下值：
DAYS
HOURS
MINUTES
SECONDS
MILLISECONDS
MICROSECONDS
NANOSECONDS
Delayed 接口也继承了 java.lang.Comparable 接口，这也就意味着 Delayed 对象之间可以进行对比。这个可能在对 DelayQueue 队列中的元素进行排序时有用，因此它们可以根据过期时间进行有序释放。
ConcurrentMap
ConcurrentMap 接口表示了一个能够对别人的访问(插入和提取)进行并发处理的 java.util.Map。
ConcurrentMap 除了从其父接口 java.util.Map 继承来的方法之外还有一些额外的原子性方法。
ConcurrentHashMap
ConcurrentMap的实现类， ConcurrentHashMap 能够提供比 HashTable 更好的并发性能。在你从中读取对象的时候 ConcurrentHashMap 并不会把整个 Map 锁住。此外，在你向其中写入对象的时候，ConcurrentHashMap 也不会锁住整个 Map。它的内部只是把 Map 中正在被写入的部分进行锁定。
另外一个不同点是，在被遍历的时候，即使是 ConcurrentHashMap 被改动，它也不会抛 ConcurrentModificationException。
CountDownLatch
CountDownLatch是一个计数器闭锁，通过它可以完成类似于阻塞当前线程的功能，即：一个线程或多个线程一直等待，直到其他线程执行的操作完成。
CountDownLatch用一个给定的计数器来初始化，该计数器的操作是原子操作，即同时只能有一个线程去操作该计数器。调用该类await方法的线程会一直处于阻塞状态，直到其他线程调用countDown方法使当前计数器的值变为零，每次调用countDown计数器的值减1。当计数器值减至零时，所有因调用await()方法而处于等待状态的线程就会继续往下执行。
计数器的初始值是线程的数量。每当一个线程执行完毕后，计数器的值就-1，当计数器的值为0时，表示所有线程都执行完毕，然后在闭锁上等待的线程就可以恢复工作了。
在某些业务场景中，程序执行需要等待某个条件完成后才能继续执行后续的操作；典型的应用如并行计算，当某个处理的运算量很大时，可以将该运算任务拆分成多个子任务，等待所有的子任务都完成之后，父任务再拿到所有子任务的运算结果进行汇总。
java.util.concurrent.atomic
java.util.concurrent.atomic原子操作类包,提供了一组原子变量类。提供具有原子性操作的方法；
其基本的特性就是在多线程环境下，当有多个线程同时执行这些类的实例包含的方法时，具有排他性，即当某个线程进入方法，执行其中的指令时，不会被其他线程打断，而别的线程就像自旋锁一样，一直等到该方法执行完成，才由JVM从等待队列中选择一个另一个线程进入，这只是一种逻辑上的理解。实际上是借助硬件的相关指令来实现的，不会阻塞线程(或者说只是在硬件级别上阻塞了)。可以对基本数据、数组中的基本数据、对类中的基本数据进行操作。原子变量类相当于一种泛化的volatile变量，能够支持原子的和有条件的读-改-写操作。
java.util.concurrent.atomic中的类可以分成4组：
标量类：AtomicBoolean，AtomicInteger，AtomicLong，AtomicReference
数组类：AtomicIntegerArray，AtomicLongArray，AtomicReferenceArray
更新器类：AtomicLongFieldUpdater，AtomicIntegerFieldUpdater，AtomicReferenceFieldUpdater
复合变量类：AtomicMarkableReference，AtomicStampedReference
一、标量类：
AtomicBoolean，AtomicInteger，AtomicLong，AtomicReference
AtomicBoolean、AtomicInteger、AtomicLong、AtomicReference这四种基本类型用来处理布尔，整数，长整数，对象四种数据，其内部实现不是简单的使用synchronized，而是一个更为高效的方式CAS (compare and swap) + volatile和native方法，从而避免了synchronized的高开销，执行效率大为提升。其实例各自提供对相应类型单个变量的访问和更新。每个类也为该类型提供适当的实用工具方法。
1、set()和get()方法可以原子的设定和获取atomic的数据，类似于volatile，保证数据会在主存中设置或读取。
2、void set()和void lazySet()：set设置为给定值，直接修改原始值；lazySet延时设置变量值，这个等价于set()方法，但是由于字段是volatile类型的，因此次字段的修改会比普通字段（非volatile字段）有稍微的性能延时（尽管可以忽略），所以如果不是想立即读取设置的新值，允许在“后台”修改值，那么此方法就很有用。
3、getAndSet()方法
原子的将变量设定为新数据，同时返回先前的旧数据。其本质是get()操作，然后做set()操作。尽管这2个操作都是atomic，但是他们合并在一起的时候，就不是atomic。在Java的源程序的级别上，如果不依赖synchronized的机制来完成这个工作，是不可能的。只有依靠native方法才可以。
4、compareAndSet()和weakCompareAndSet()
这两个方法都是conditional modifier方法。这2个方法接受2个参数，一个是期望数据(expected)，一个是新数据(new)；如果atomic里面的数据和期望数据一 致，则将新数据设定给atomic的数据，返回true，表明成功；否则就不设定，并返回false。JDK规范中说：以原子方式读取和有条件地写入变量但不 创建任何 happen-before 排序，因此不提供与除 weakCompareAndSet 目标外任何变量以前或后续读取或写入操作有关的任何保证。大意就是说调用weakCompareAndSet时并不能保证不存在happen- before的发生（也就是可能存在指令重排序导致此操作失败）。但是从Java源码来看，其实此方法并没有实现JDK规范的要求，最后效果和 compareAndSet是等效的，都调用了unsafe.compareAndSwapInt()完成操作。
虽然原子的标量类扩展了基本类型的类，但是并没有扩展基本类型的包装类，如Integer或Long，事实上它们也不能直接扩展。因为基本类型的包装类是不可以修改的，而原子变量类是可以修改的。在原子变量类中没有重新定义hashCode或equals方法，每个实例都是不同的，他们也不宜用做基于散列容器中的键值。
二、数组类：
AtomicIntegerArray，AtomicLongArray，AtomicReferenceArray
AtomicIntegerArray、AtomicLongArray、AtomicReferenceArray类进一步扩展了原子操作，对这些类型的数组提供了支持。这些类在为其数组元素提供volatile访问语义方面也引人注目，这对于普通数组来说是不受支持的。其内部并不是像AtomicInteger一样维持一个volatile变量，而是全部由native方法实现。数组变量进行volatile没有意义，因此set/get就需要unsafe来做了，但是多了一个index来指定操作数组中的哪一个元素。
三、更新器类：
AtomicLongFieldUpdater，AtomicIntegerFieldUpdater，AtomicReferenceFieldUpdater
AtomicReferenceFieldUpdater,AtomicIntegerFieldUpdater和AtomicLongFieldUpdater 是基于反射的实用工具，可以提供对关联字段类型的访问，可用于获取任意选定volatile字段上的compareAndSet操作。它们主要用于原子数据结构中，该结构中同一节点的几个 volatile 字段都独立受原子更新控制。这些类在如何以及何时使用原子更新方面具有更大的灵活性，但相应的弊端是基于映射的设置较为拙笨、使用不太方便，而且在保证方面也较差。
使用中要注意一下几点：
（1）字段必须是volatile类型的
（2）字段的描述类型（修饰符public/protected/default/private）是与调用者与操作对象字段的关系一致。也就是说 调用者能够直接操作对象字段，那么就可以反射进行原子操作。但是对于父类的字段，子类是不能直接操作的，尽管子类可以访问父类的字段。
（3）只能是实例变量，不能是类变量，也就是说不能加static关键字。
（4）只能是可修改变量，不能使final变量，因为final的语义就是不可修改。实际上final的语义和volatile是有冲突的，这两个关键字不能同时存在。
（5）对于AtomicIntegerFieldUpdater 和AtomicLongFieldUpdater 只能修改int/long类型的字段，不能修改其包装类型（Integer/Long）。如果要修改包装类型就需要使用AtomicReferenceFieldUpdater 。
四、复合变量类：
AtomicMarkableReference，AtomicStampedReference
AtomicMarkableReference 类将单个布尔值与引用关联起来。维护带有标记位的对象引用，可以原子方式更新带有标记位的引用类型。
AtomicStampedReference 类将整数值与引用关联起来。维护带有整数“标志”的对象引用，可以原子更新带有版本号的引用类型。该类将整数值与引用关联起来，可用于原子的更新数据和版本号，可以解决使用CAS进行原子更新时，可能出现的ABA问题。
附录：java并发编程中的CAS和ABA问题
CAS(compare and swap)比较和替换是java5+提供的设计并发算法时用到的一种技术。是使用一个期望值和一个变量的当前值进行比较，如果当前值和我们的期望值相等，就使用一个新值替换掉当前值。
java.util.concurrent.locks
Lock
Lock接口：
下面来逐个讲述Lock接口中每个方法的使用，lock()、tryLock()、tryLock(long time, TimeUnit unit)和lockInterruptibly()是用来获取锁的。unLock()方法是用来释放锁的。newCondition()这个方法放在最后讲。
在Lock中声明了四个方法来获取锁，那么这四个方法有何区别呢？
lock ： 在锁上等待，直到获取锁；
tryLock：立即返回，获得锁返回true,没获得锁返回false；
tryInterruptibly：在锁上等待，直到获取锁，但是会响应中断，这个方法优先考虑响应中断，而不是响应锁的普通获取或重入获取。
首先lock()方法是平常使用得最多的一个方法，就是用来获取锁。如果锁已被其他线程获取，则进行等待。
由于在前面讲到如果采用Lock，必须主动去释放锁，并且在发生异常时，不会自动释放锁。因此一般来说，使用Lock必须在try{}catch{}块中进行，并且将释放锁的操作放在finally块中进行，以保证锁一定被被释放，防止死锁的发生。通常使用Lock来进行同步的话，是以下面这种形式去使用的
Lock lock = ...;
lock.lock();
try{
//处理任务
}catch(Exception ex){
}finally{
lock.unlock(); //释放锁
}
tryLock()方法是有返回值的，它表示用来尝试获取锁，如果获取成功，则返回true，如果获取失败（即锁已被其他线程获取），则返回false，也就说这个方法无论如何都会立即返回。在拿不到锁时不会一直在那等待。
tryLock(long time, TimeUnit unit)方法和tryLock()方法是类似的，只不过区别在于这个方法在拿不到锁时会等待一定的时间，在时间期限之内如果还拿不到锁，就返回false。如果如果一开始拿到锁或者在等待期间内拿到了锁，则返回true。
所以，一般情况下通过tryLock来获取锁时是这样使用的：
Lock lock = ...;
if(lock.tryLock()) {
try{
//处理任务
}catch(Exception ex){
}finally{
lock.unlock(); //释放锁
}
}else {
//如果不能获取锁，则直接做其他事情
}ockInterruptibly()方法比较特殊，当通过这个方法去获取锁时，如果线程正在等待获取锁，则这个线程能够响应中断，即中断线程的等待状态。也就使说，当两个线程同时通过lock.lockInterruptibly()想获取某个锁时，假若此时线程A获取到了锁，而线程B只有在等待，那么对线程B调用threadB.interrupt()方法能够中断线程B的等待过程。
由于lockInterruptibly()的声明中抛出了异常，所以lock.lockInterruptibly()必须放在try块中或者在调用lockInterruptibly()的方法外声明抛出InterruptedException。
因此lockInterruptibly()一般的使用形式如下：
|
public void method() throws InterruptedException { lock.lockInterruptibly(); try { //..... } finally { lock.unlock(); } } |
注意，当一个线程获取了锁之后，是不会被interrupt()方法中断的。因为本身在前面的文章中讲过单独调用interrupt()方法不能中断正在运行过程中的线程，只能中断阻塞过程中的线程。
因此当通过lockInterruptibly()方法获取某个锁时，如果不能获取到，只有进行等待的情况下，是可以响应中断的。
而用synchronized修饰的话，当一个线程处于等待某个锁的状态，是无法被中断的，只有一直等待下去。
ReentrantLock
意思是“可重入锁”，
ReentrantLock是唯一实现了Lock接口的类，并且ReentrantLock提供了更多的方法。
使用方式：
|
public class Test { private ArrayList<Integer> arrayList = new ArrayList<Integer>(); public static void main(String[] args) { final Test test = new Test(); new Thread(){ public void run() { test.insert(Thread.currentThread()); }; }.start();
new Thread(){ public void run() { test.insert(Thread.currentThread()); }; }.start(); }
public void insert(Thread thread) { Lock lock = new ReentrantLock(); //注意这个地方 lock.lock(); try { System.out.println(thread.getName()+"得到了锁"); for(int i=0;i<5;i++) { arrayList.add(i); } } catch (Exception e) { // TODO: handle exception }finally { System.out.println(thread.getName()+"释放了锁"); lock.unlock(); } } } |
这样如果执行的话
|
Thread-0得到了锁 Thread-1得到了锁 Thread-0释放了锁 Thread-1释放了锁 |
在insert方法中的lock变量是局部变量，每个线程执行该方法时都会new一个新的ReentrantLock，那么理所当然每个线程执行到lock.lock()处获取的是不同的锁，所以就不会发生冲突。
只需要将lock声明为类的属性即可。
|
public class Test { private ArrayList<Integer> arrayList = new ArrayList<Integer>(); private Lock lock = new ReentrantLock(); //注意这个地方 public static void main(String[] args) { final Test test = new Test();
new Thread(){ public void run() { test.insert(Thread.currentThread()); }; }.start();
new Thread(){ public void run() { test.insert(Thread.currentThread()); }; }.start(); }
public void insert(Thread thread) { lock.lock(); try { System.out.println(thread.getName()+"得到了锁"); for(int i=0;i<5;i++) { arrayList.add(i); } } catch (Exception e) { // TODO: handle exception }finally { System.out.println(thread.getName()+"释放了锁"); lock.unlock(); } } } |
tryLock()的使用方法
|
public class Test { private ArrayList<Integer> arrayList = new ArrayList<Integer>(); private Lock lock = new ReentrantLock(); //注意这个地方 public static void main(String[] args) { final Test test = new Test();
new Thread(){ public void run() { test.insert(Thread.currentThread()); }; }.start();
new Thread(){ public void run() { test.insert(Thread.currentThread()); }; }.start(); }
public void insert(Thread thread) { if(lock.tryLock()) { try { System.out.println(thread.getName()+"得到了锁"); for(int i=0;i<5;i++) { arrayList.add(i); } } catch (Exception e) { // TODO: handle exception }finally { System.out.println(thread.getName()+"释放了锁"); lock.unlock(); } } else { System.out.println(thread.getName()+"获取锁失败"); } } } |
输出结果
|
Thread-0得到了锁 Thread-1获取锁失败 Thread-0释放了锁 |
lockInterruptibly()响应中断的使用方法：
|
public class Test { private Lock lock = new ReentrantLock(); public static void main(String[] args) { Test test = new Test(); MyThread thread1 = new MyThread(test); MyThread thread2 = new MyThread(test); thread1.start(); thread2.start();
try { Thread.sleep(2000); } catch (InterruptedException e) { e.printStackTrace(); } thread2.interrupt(); }
public void insert(Thread thread) throws InterruptedException{ lock.lockInterruptibly(); //注意，如果需要正确中断等待锁的线程，必须将获取锁放在外面，然后将InterruptedException抛出 try { System.out.println(thread.getName()+"得到了锁"); Thread.sleep(5000); } finally { System.out.println(Thread.currentThread().getName()+"执行finally"); lock.unlock(); System.out.println(thread.getName()+"释放了锁"); } } }
class MyThread extends Thread { private Test test = null; public MyThread(Test test) { this.test = test; } @Override public void run() {
try { test.insert(Thread.currentThread()); } catch (InterruptedException e) { System.out.println(Thread.currentThread().getName()+"被中断"); } } } |
运行之后，发现thread2能够被正确中断。结果：
Thread-0得到了锁
Thread-1被中断
Thread-0执行finally
Thread-0释放了锁
同样的代码如果使用lock.lock()，结果如下：
Thread-0得到了锁
Thread-0执行finally
Thread-0释放了锁
Thread-1得到了锁
Thread-1执行finally
Thread-1释放了锁
Thread-1被中断
因为lock.lock()不会相应中断。
ReadWriteLock
ReadWriteLock也是一个接口，在它里面只定义了两个方法：
Lock readLock();
Lock writeLock();
一个用来获取读锁，一个用来获取写锁。也就是说将文件的读写操作分开，分成2个锁来分配给线程，从而使得多个线程可以同时进行读操作。
ReentrantReadWriteLock
实现了ReadWriteLock接口，最主要的有两个方法：readLock()和writeLock()用来获取读锁和写锁。
示例：
|
public class Test { private ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();
public static void main(String[] args) { final Test test = new Test();
new Thread(){ public void run() { test.get(Thread.currentThread()); }; }.start();
new Thread(){ public void run() { test.get(Thread.currentThread()); }; }.start();
}
public void get(Thread thread) { rwl.readLock().lock(); try { long start = System.currentTimeMillis();
while(System.currentTimeMillis() - start <= 1) { System.out.println(thread.getName()+"正在进行读操作"); } System.out.println(thread.getName()+"读操作完毕"); } finally { rwl.readLock().unlock(); } } } |
此时打印的结果为：
|
Thread-0正在进行读操作 Thread-0正在进行读操作 Thread-1正在进行读操作 Thread-0正在进行读操作 Thread-1正在进行读操作 Thread-0正在进行读操作 Thread-1正在进行读操作 Thread-1正在进行读操作 Thread-1正在进行读操作 Thread-1正在进行读操作 Thread-1正在进行读操作 Thread-1正在进行读操作 Thread-0正在进行读操作 Thread-0正在进行读操作 Thread-0正在进行读操作 Thread-0正在进行读操作 Thread-1正在进行读操作 Thread-1正在进行读操作 Thread-1正在进行读操作 Thread-1正在进行读操作 Thread-0正在进行读操作 Thread-0读操作完毕 Thread-1读操作完毕 |
说明thread1和thread2在同时进行读操作。
这样就大大提升了读操作的效率。
不过要注意的是，如果有一个线程已经占用了读锁，则此时其他线程如果要申请写锁，则申请写锁的线程会一直等待释放读锁。
如果有一个线程已经占用了写锁，则此时其他线程如果申请写锁或者读锁，则申请的线程会一直等待释放写锁。
java.util.concurrent.Callable
Callable可以理解为一个带返回值的多线程接口
Callable和Runnable有几点不同：
（1）Callable规定的方法是call()，而Runnable规定的方法是run().
（2）Callable的任务执行后可返回值，而Runnable的任务是不能返回值的。
（3）call()方法可抛出异常，而run()方法是不能抛出异常的。
（4）运行Callable任务可拿到一个Future对象，可了解任务执行情况，可取消任务的执行，还可获取任务执行的结果。
java.util.concurrent.Future
文章引用说明：https://www.cnblogs.com/ludongguoa/p/15316488.html
Future是Java5新加的接口，提供了一种异步并行计算的功能。如果主线程需要执行一个很耗时的计算任务，我们就可以通过future把这个任务放到异步线程中执行。主线程继续处理其他任务，处理完成后，再通过Future获取计算结果。
使用示例：
public class UserInfoService {
public UserInfo getUserInfo(Long userId) throws InterruptedException {
Thread.sleep(300);//模拟调用耗时
return new UserInfo("666", "捡田螺的小男孩", 27);
}
}
public class MedalService {
public MedalInfo getMedalInfo(long userId) throws InterruptedException {
Thread.sleep(500); //模拟调用耗时
return new MedalInfo("666", "守护勋章");
}
}public class FutureTest {
public static void main(String[] args) throws ExecutionException, InterruptedException {
ExecutorService executorService = Executors.newFixedThreadPool(10);
UserInfoService userInfoService = new UserInfoService();
MedalService medalService = new MedalService();
long userId =666L;
long startTime = System.currentTimeMillis();
//调用用户服务获取用户基本信息
FutureTask<UserInfo> userInfoFutureTask = new FutureTask<>(new Callable<UserInfo>() {
@Override
public UserInfo call() throws Exception {
return userInfoService.getUserInfo(userId);
}
});
executorService.submit(userInfoFutureTask);
Thread.sleep(300); //模拟主线程其它操作耗时
FutureTask<MedalInfo> medalInfoFutureTask = new FutureTask<>(new Callable<MedalInfo>() {
@Override
public MedalInfo call() throws Exception {
return medalService.getMedalInfo(userId);
}
});
executorService.submit(medalInfoFutureTask);
UserInfo userInfo = userInfoFutureTask.get();//获取个人信息结果
MedalInfo medalInfo = medalInfoFutureTask.get();//获取勋章信息结果
System.out.println("总共用时" + (System.currentTimeMillis() - startTime) + "ms");
}
}
get()方法会阻塞主线程，在get方法没有拿到返回值前会一直阻塞。
Future提供了一个isDone方法，可以在程序中轮询这个方法查询执行结果。阻塞的方式和异步编程的设计理念相违背，而轮询的方式会耗费无谓的CPU资源。因此，JDK8设计出CompletableFuture。
java.util.concurrent.CompletableFuture
Runnable+Thread虽然提供了多线程的能力但是没有返回值。
Callable+Thread的方法提供多线程和返回值的能力但是在获取返回值的时候会阻塞主线程。
CompletableFuture的应用场景：
示例：
public class FutureTest {
public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {
UserInfoService userInfoService = new UserInfoService();
MedalService medalService = new MedalService();
long userId =666L;
long startTime = System.currentTimeMillis();
//调用用户服务获取用户基本信息
CompletableFuture<UserInfo> completableUserInfoFuture = CompletableFuture.supplyAsync(() -> userInfoService.getUserInfo(userId));
Thread.sleep(300); //模拟主线程其它操作耗时
CompletableFuture<MedalInfo> completableMedalInfoFuture = CompletableFuture.supplyAsync(() -> medalService.getMedalInfo(userId));
UserInfo userInfo = completableUserInfoFuture.get(2,TimeUnit.SECONDS);//获取个人信息结果
MedalInfo medalInfo = completableMedalInfoFuture.get();//获取勋章信息结果
System.out.println("总共用时" + (System.currentTimeMillis() - startTime) + "ms");
}
}get()方法是阻塞的，可以调用get方法时指定超时时间
线程池说明
如果不指定线程池默认使用的内置线程池ForkJoinPool.commonPool()，根据supplier构建执行任务，适用于计算密集型任务，尤其是那些可以被细分为更小的子任务并行处理的任务。
默认线下池的线程个数是CPU核数-1，对于一个 4 核的机器来说，最只有 3 个线程，对于IO密集型的任务来说，这其实远远不够用，从而导致大量的IO任务在等待，甚至服务挂掉。
（计算密集型任务的特点是要进行大量的计算，消耗CPU资源；IO密集型，涉及到网络、磁盘IO的任务都是IO密集型任务，这类任务的特点是CPU消耗很少，任务的大部分时间都在等待IO操作完成（因为IO的速度远远低于CPU和内存的速度）。对于IO密集型任务，任务越多，CPU效率越高，但也有一个限度。常见的大部分任务都是IO密集型任务，比如Web应用。）
io密集型的使用ThreadPoolExecutor 进行自定义线程池就可以。
commonPool 是当前机器进程上的所有 CompletableFuture、parallelStream（并行流） 共享的。
当CPU核心数-1>1时，才会使用默认的线程池，否则将会为每个CompletableFuture的任务创建一个新线程去执行。如果不是双核及以上的服务器就会有资源耗尽的风险
ForkJoinPool线程池最大的特点就是分叉(fork)合并(join)，将一个大任务拆分成多个小任务，并行执行，再结合工作窃取模式(worksteal)提高整体的执行效率，充分利用CPU资源。
自定义线程池使用：
@Configuration
public class ThreadPoolConfig {
//参数初始化
private static final int CPU_COUNT = Runtime.getRuntime().availableProcessors();
//核心线程数量大小
private static final int corePoolSize = Math.max(2, Math.min(CPU_COUNT-1,4));
//线程池最大容纳线程数
private static final int maxPoolSize = CPU_COUNT * 2 + 1;
//阻塞队列
private static final int workQueue = 20;
//线程空闲后的存活时长
private static final int keepAliveTime = 30;
@Bean("asyncTaskExecutor")
public Executor getAsyncExecutor() {
ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
//核心线程数
threadPoolTaskExecutor.setCorePoolSize(corePoolSize);
//最大线程数
threadPoolTaskExecutor.setMaxPoolSize(maxPoolSize);
//等待队列
threadPoolTaskExecutor.setQueueCapacity(workQueue);
//线程前缀
threadPoolTaskExecutor.setThreadNamePrefix("asyncTaskExecutor-");
//线程池维护线程所允许的空闲时间,单位为秒
threadPoolTaskExecutor.setKeepAliveSeconds(keepAliveTime);
// 线程池对拒绝任务(无线程可用)的处理策略
threadPoolTaskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
threadPoolTaskExecutor.initialize();
return threadPoolTaskExecutor;
}
}@RestController
@RequestMapping("/task")
public class CompletableTaskController {
private final Logger logger = LoggerFactory.getLogger(this.getClass());
@Autowired
@Qualifier("asyncTaskExecutor")
private Executor asyncTaskExecutor;
@RequestMapping("testOrderTask")
public String testOrderTask(){
List<CompletableFuture<List<Integer>>> futureList = Lists.newArrayList();
// 任务1，计算3秒
CompletableFuture<List<Integer>> task1 = CompletableFuture.supplyAsync(() -> {
sleepSeconds(3L);
return Lists.newArrayList(1,2,3);
}, asyncTaskExecutor);
futureList.add(task1);
// 任务2，计算2秒，得答案5
CompletableFuture<List<Integer>> task2 = CompletableFuture.supplyAsync(() -> {
sleepSeconds(1L);
return Lists.newArrayList(4,5,6);
}, asyncTaskExecutor);
futureList.add(task2);
// 任务3，计算3秒，得答案5
CompletableFuture<List<Integer>> task3 = CompletableFuture.supplyAsync(() -> {
sleepSeconds(2L);
return Lists.newArrayList(7,8,9);
}, asyncTaskExecutor);
futureList.add(task3);
// 写法1
List<Integer> newList = futureList.stream().map(CompletableFuture::join).flatMap(List::stream).collect(Collectors.toList());
return JSON.toJSONString(newList);
}
private void sleepSeconds(long seconds) {
try {
Thread.sleep(seconds * 1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}supplyAsync,runAsync
supplyAsync，支持返回值
runAsync执行，没有返回值。
//使用默认内置线程池ForkJoinPool.commonPool()，根据supplier构建执行任务，默认线下池的线程个数是CPU核数-1
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
//自定义线程，根据supplier构建执行任务
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)//使用默认内置线程池ForkJoinPool.commonPool()，根据runnable构建执行任务
public static CompletableFuture<Void> runAsync(Runnable runnable)
//自定义线程，根据runnable构建执行任务
public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor)public class FutureTest {
public static void main(String[] args) {
//可以自定义线程池
ExecutorService executor = Executors.newCachedThreadPool();
//runAsync的使用
CompletableFuture<Void> runFuture = CompletableFuture.runAsync(() -> System.out.println("run,关注公众号:捡田螺的小男孩"), executor);
//supplyAsync的使用
CompletableFuture<String> supplyFuture = CompletableFuture.supplyAsync(() -> {
System.out.print("supply,关注公众号:捡田螺的小男孩");
return "捡田螺的小男孩"; }, executor);
//runAsync的future没有返回值，输出null
System.out.println(runFuture.join());
//supplyAsync的future，有返回值
System.out.println(supplyFuture.join());
executor.shutdown(); // 线程池需要关闭
}
}
//输出
run,关注公众号:捡田螺的小男孩
null
supply,关注公众号:捡田螺的小男孩捡田螺的小男孩thenRun,thenRunAsync
public CompletableFuture<Void> thenRun(Runnable action);
public CompletableFuture<Void> thenRunAsync(Runnable action);thenRun方法，做完第一个任务后，再做第二个任务。某个任务执行完成后，执行回调方法；但是前后两个任务没有参数传递，第二个任务也没有返回值
public class FutureThenRunTest {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture<String> orgFuture = CompletableFuture.supplyAsync(
()->{
System.out.println("先执行第一个CompletableFuture方法任务");
return "捡田螺的小男孩";
}
);
CompletableFuture thenRunFuture = orgFuture.thenRun(() -> {
System.out.println("接着执行第二个任务");
});
System.out.println(thenRunFuture.get());
}
}
//输出
先执行第一个CompletableFuture方法任务
接着执行第二个任务
nullthenRunAsync
private static final Executor asyncPool = useCommonPool ?
ForkJoinPool.commonPool() : new ThreadPerTaskExecutor();
public CompletableFuture<Void> thenRun(Runnable action) {
return uniRunStage(null, action);
}
public CompletableFuture<Void> thenRunAsync(Runnable action) {
return uniRunStage(asyncPool, action);
}如果你执行第一个任务的时候，传入了一个自定义线程池：
调用thenRun方法执行第二个任务时，则第二个任务和第一个任务是共用同一个线程池。
调用thenRunAsync执行第二个任务时，则第一个任务使用的是你自己传入的线程池，第二个任务使用的是ForkJoin线程池
thenAccept,thenAcceptAsync
thenAccept方法表示，第一个任务执行完成后，执行第二个回调方法任务，会将该任务的执行结果，作为入参，传递到回调方法中，但是回调方法是没有返回值的。
public class FutureThenAcceptTest {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture<String> orgFuture = CompletableFuture.supplyAsync(
()->{
System.out.println("原始CompletableFuture方法任务");
return "捡田螺的小男孩";
}
);
CompletableFuture thenAcceptFuture = orgFuture.thenAccept((a) -> {
if ("捡田螺的小男孩".equals(a)) {
System.out.println("关注了");
}
System.out.println("先考虑考虑");
});
System.out.println(thenAcceptFuture.get());
}
}thenApply,thenApplyAsync
thenApply方法表示，第一个任务执行完成后，执行第二个回调方法任务，会将该任务的执行结果，作为入参，传递到回调方法中，并且回调方法是有返回值的。
public class FutureThenApplyTest {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture<String> orgFuture = CompletableFuture.supplyAsync(
()->{
System.out.println("原始CompletableFuture方法任务");
return "捡田螺的小男孩";
}
);
CompletableFuture<String> thenApplyFuture = orgFuture.thenApply((a) -> {
if ("捡田螺的小男孩".equals(a)) {
return "关注了";
}
return "先考虑考虑";
});
System.out.println(thenApplyFuture.get());
}
}
//输出
原始CompletableFuture方法任务
关注了exceptionally
exceptionally方法表示，某个任务执行异常时，执行的回调方法;并且有抛出异常作为参数，传递到回调方法。
public class FutureExceptionTest {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture<String> orgFuture = CompletableFuture.supplyAsync(
()->{
System.out.println("当前线程名称：" + Thread.currentThread().getName());
throw new RuntimeException();
}
);
CompletableFuture<String> exceptionFuture = orgFuture.exceptionally((e) -> {
e.printStackTrace();
return "你的程序异常啦";
});
System.out.println(exceptionFuture.get());
}
}
//输出
当前线程名称：ForkJoinPool.commonPool-worker-1
java.util.concurrent.CompletionException: java.lang.RuntimeException
at java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:273)
at java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:280)
at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1592)
at java.util.concurrent.CompletableFuture$AsyncSupply.exec(CompletableFuture.java:1582)
at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)
at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056)
at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692)
at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:157)
Caused by: java.lang.RuntimeException
at cn.eovie.future.FutureWhenTest.lambda$main$0(FutureWhenTest.java:13)
at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1590)
... 5 more
你的程序异常啦whenComplete
whenComplete方法表示，某个任务执行完成后，执行的回调方法，无返回值；并且whenComplete方法返回的CompletableFuture的result是上个任务的结果。
public class FutureWhenTest {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture<String> orgFuture = CompletableFuture.supplyAsync(
()->{
System.out.println("当前线程名称：" + Thread.currentThread().getName());
try {
Thread.sleep(2000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
return "捡田螺的小男孩";
}
);
CompletableFuture<String> rstFuture = orgFuture.whenComplete((a, throwable) -> {
System.out.println("当前线程名称：" + Thread.currentThread().getName());
System.out.println("上个任务执行完啦，还把" + a + "传过来");
if ("捡田螺的小男孩".equals(a)) {
System.out.println("666");
}
System.out.println("233333");
});
System.out.println(rstFuture.get());
}
}
//输出
当前线程名称：ForkJoinPool.commonPool-worker-1
当前线程名称：ForkJoinPool.commonPool-worker-1
上个任务执行完啦，还把捡田螺的小男孩传过来
666
233333
捡田螺的小男孩handle
handle方法表示，任务执行完成后，执行回调方法，并且是有返回值的;并且handle方法返回的CompletableFuture的result是回调方法执行的结果。
public class FutureHandlerTest {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture<String> orgFuture = CompletableFuture.supplyAsync(
()->{
System.out.println("当前线程名称：" + Thread.currentThread().getName());
try {
Thread.sleep(2000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
return "捡田螺的小男孩";
}
);
CompletableFuture<String> rstFuture = orgFuture.handle((a, throwable) -> {
System.out.println("上个任务执行完啦，还把" + a + "传过来");
if ("捡田螺的小男孩".equals(a)) {
System.out.println("666");
return "关注了";
}
System.out.println("233333");
return null;
});
System.out.println(rstFuture.get());
}
}
//输出
当前线程名称：ForkJoinPool.commonPool-worker-1
上个任务执行完啦，还把捡田螺的小男孩传过来
666
关注了thenCombine / thenAcceptBoth / runAfterBoth
都表示：
将两个CompletableFuture组合起来，只有这两个都正常执行完了，才会执行某个任务。
thenCombine：会将两个任务的执行结果作为方法入参，传递到指定方法中，且有返回值
thenAcceptBoth: 会将两个任务的执行结果作为方法入参，传递到指定方法中，且无返回值
runAfterBoth 不会把执行结果当做方法入参，且没有返回值。
public class ThenCombineTest {
public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {
CompletableFuture<String> first = CompletableFuture.completedFuture("第一个异步任务");
ExecutorService executor = Executors.newFixedThreadPool(10);
CompletableFuture<String> future = CompletableFuture
//第二个异步任务
.supplyAsync(() -> "第二个异步任务", executor)
// (w, s) -> System.out.println(s) 是第三个任务
.thenCombineAsync(first, (s, w) -> {
System.out.println(w);
System.out.println(s);
return "两个异步任务的组合";
}, executor);
System.out.println(future.join());
executor.shutdown();
}
}
//输出
第一个异步任务
第二个异步任务
两个异步任务的组合applyToEither / acceptEither / runAfterEither
都表示：
将两个CompletableFuture组合起来，只要其中一个执行完了,就会执行某个任务。
区别在于：
applyToEither：会将已经执行完成的任务，作为方法入参，传递到指定方法中，且有返回值
acceptEither: 会将已经执行完成的任务，作为方法入参，传递到指定方法中，且无返回值
runAfterEither：不会把执行结果当做方法入参，且没有返回值。
public class AcceptEitherTest {
public static void main(String[] args) {
//第一个异步任务，休眠2秒，保证它执行晚点
CompletableFuture<String> first = CompletableFuture.supplyAsync(()->{
try{
Thread.sleep(2000L);
System.out.println("执行完第一个异步任务");}
catch (Exception e){
return "第一个任务异常";
}
return "第一个异步任务";
});
ExecutorService executor = Executors.newSingleThreadExecutor();
CompletableFuture<Void> future = CompletableFuture
//第二个异步任务
.supplyAsync(() -> {
System.out.println("执行完第二个任务");
return "第二个任务";}
, executor)
//第三个任务
.acceptEitherAsync(first, System.out::println, executor);
executor.shutdown();
}
}
//输出
执行完第二个任务
第二个任务import java.util.concurrent.CompletableFuture;
import java.util.stream.IntStream;
public class CompletableFutureExample {
public static void main(String[] args) {
// 创建一个 CompletableFuture 数组或列表，用于存储每个任务的 CompletableFuture 对象
CompletableFuture<Void>[] futures = new CompletableFuture[10];
// 使用 IntStream.range 来模拟 for 循环
IntStream.range(0, 10).forEach(i -> {
// 每次迭代创建并启动一个新任务
futures[i] = CompletableFuture.runAsync(() -> {
// 这里是你的任务代码
System.out.println("Task " + i + " is running on thread: " + Thread.currentThread().getName());
try {
Thread.sleep(1000); // 模拟耗时操作
} catch (InterruptedException e) {
e.printStackTrace();
}
});
});
// 使用 CompletableFuture.allOf 方法来等待所有任务完成
CompletableFuture.allOf(futures).thenRun(() -> {
// 所有任务完成后执行的代码
System.out.println("All tasks are completed.");
});
}
}AllOf
所有任务都执行完成后，才执行 allOf返回的CompletableFuture。如果任意一个任务异常，allOf的CompletableFuture，执行get方法，会抛出异常
public class allOfFutureTest {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture<Void> a = CompletableFuture.runAsync(()->{
System.out.println("我执行完了");
});
CompletableFuture<Void> b = CompletableFuture.runAsync(() -> {
System.out.println("我也执行完了");
});
CompletableFuture<Void> allOfFuture = CompletableFuture.allOf(a, b).whenComplete((m,k)->{
System.out.println("finish");
});
}
}
//输出
我执行完了
我也执行完了
finishAnyOf
任意一个任务执行完，就执行anyOf返回的CompletableFuture。如果执行的任务异常，anyOf的CompletableFuture，执行get方法，会抛出异常
public class AnyOfFutureTest {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture<Void> a = CompletableFuture.runAsync(()->{
try {
Thread.sleep(3000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("我执行完了");
});
CompletableFuture<Void> b = CompletableFuture.runAsync(() -> {
System.out.println("我也执行完了");
});
CompletableFuture<Object> anyOfFuture = CompletableFuture.anyOf(a, b).whenComplete((m,k)->{
System.out.println("finish");
// return "捡田螺的小男孩";
});
anyOfFuture.join();
}
}
//输出
我也执行完了
finishthenCompose
thenCompose方法会在某个任务执行完成后，将该任务的执行结果,作为方法入参,去执行指定的方法。该方法会返回一个新的CompletableFuture实例
如果该CompletableFuture实例的result不为null，则返回一个基于该result新的CompletableFuture实例；
如果该CompletableFuture实例为null，然后就执行这个新任务
public class ThenComposeTest {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture<String> f = CompletableFuture.completedFuture("第一个任务");
//第二个异步任务
ExecutorService executor = Executors.newSingleThreadExecutor();
CompletableFuture<String> future = CompletableFuture
.supplyAsync(() -> "第二个任务", executor)
.thenComposeAsync(data -> {
System.out.println(data); return f; //使用第一个任务作为返回
}, executor);
System.out.println(future.join());
executor.shutdown();
}
}
//输出
第二个任务
第一个任务
概念说明
可重入锁
如果锁具备可重入性，则称作为可重入锁。像synchronized和ReentrantLock都是可重入锁，当一个线程执行到某个synchronized方法时，比如说method1，而在method1中会调用另外一个synchronized方法method2，此时线程不必重新去申请锁，而是可以直接执行方法method2。
|
class MyClass { public synchronized void method1() { method2(); }
public synchronized void method2() {
} } |
上述代码中的两个方法method1和method2都用synchronized修饰了，假如某一时刻，线程A执行到了method1，此时线程A获取了这个对象的锁，而由于method2也是synchronized方法，假如synchronized不具备可重入性，此时线程A需要重新申请锁。但是这就会造成一个问题，因为线程A已经持有了该对象的锁，而又在申请获取该对象的锁，这样就会线程A一直等待永远不会获取到的锁。
而由于synchronized和Lock都具备可重入性，所以不会发生上述现象。
可中断锁
可中断锁：顾名思义，就是可以相应中断的锁。
在Java中，synchronized就不是可中断锁，而Lock是可中断锁。
如果某一线程A正在执行锁中的代码，另一线程B正在等待获取该锁，可能由于等待时间过长，线程B不想等待了，想先处理其他事情，我们可以让它中断自己或者在别的线程中中断它，这种就是可中断锁。
公平锁
公平锁即尽量以请求锁的顺序来获取锁。比如同是有多个线程在等待一个锁，当这个锁被释放时，等待时间最久的线程（最先请求的线程）会获得该所，这种就是公平锁。
非公平锁即无法保证锁的获取是按照请求锁的顺序进行的。这样就可能导致某个或者一些线程永远获取不到锁。
在Java中，synchronized就是非公平锁，它无法保证等待的线程获取锁的顺序。
而对于ReentrantLock和ReentrantReadWriteLock，它默认情况下是非公平锁，但是可以设置为公平锁。
读写锁
读写锁将对一个资源（比如文件）的访问分成了2个锁，一个读锁和一个写锁。
正因为有了读写锁，才使得多个线程之间的读操作不会发生冲突。
写锁与读锁写锁都冲突
有界队列和无界队列
有界队列：就是有固定大小的队列。比如设定了固定大小的 LinkedBlockingQueue，又或者大小为 0，只是在生产者和消费者中做中转用的 SynchronousQueue。
无界队列：指的是没有设置固定大小的队列。这些队列的特点是可以直接入列，直到溢出。当然现实几乎不会有到这么大的容量（超过 Integer.MAX_VALUE），所以从使用者的体验上，就相当于 “无界”。比如没有设定固定大小的 LinkedBlockingQueue。
常见的并发机制
信号量：用于进程间传递信号的一个整数值。在信号量上只有三种操作可以进行：初始化、递减和增加，这三种都是原子操作。递减操作可以用于阻塞一个进程，增加操作可以用于解除阻塞一个进程。也称为计数信号量或一般信号量。
二元信号量：只取0值和1值的信号量
互斥量：类似于二元信号量。关键区别在于为其加锁（设定值为0）的进程和为其解锁（设定值为1）的进程必须为同一个进程
管程：一种编程语言结构，在一个抽象数据类型中封装了变量、访问过程和初始化代码。管程的变量只能由管程自己的访问过程，每次只能有一个进程在其中执行。访问过程即临界区。管程可以有一个等待进程队列。
事件标志：作为同步机制的一个内存字。应用程序代码可以为标志中的每个位关联不同的时间。通过测试相关的一个或多个位，线程可以等待一个事件或多个事件。在全部的所需位都被设定至少一个位设定OR之前，线程会阻塞。
消息：两个进程交互信息的一种方法，也可以用于同步。
自旋锁：一个互斥机制，进程在一个无条件循环中执行，等待锁变量的值变成可用。
