Java并发编程指南

2022-12-17 08:20:29

　　多线程是实现并发机制的一种有效手段。在 Java 中实现多线程有两种手段，一种是继承 Thread 类，另一种就是实现 Runnable/Callable 接口。

　　java.util.concurrent 包是专为 Java并发编程而设计的包。类图如下：

一、同步

1.1 synchronized 关键字，用来给对象和方法或者代码块加锁。

同步方法

synchronized T methodName(){}

同步方法锁定的是当前对象。当多线程通过同一个对象引用多次调用当前同步方法时，需同步执行。静态同步方法，锁的是当前类型的类对象。
同步方法只影响锁定同一个锁对象的同步方法。不影响其他线程调用非同步方法，或调用其他锁资源的同步方法。
锁可重入。 同一个线程，多次调用同步代码，锁定同一个锁对象，可重入。子类同步方法覆盖父类同步方法。可以指定调用父类的同步方法，相当于锁的重入。
当同步方法中发生异常的时候，自动释放锁资源。不会影响其他线程的执行。

同步代码块

T methodName(){

    synchronized(object){}

}

同步代码块在执行时，是锁定 object 对象。当多个线程调用同一个方法时，锁定对象不变的情况下，需同步执行。

同步代码块的同步粒度更加细致，效率更高。

T methodName(){

    synchronized(this){}

}

当锁定对象为 this 时，相当于同步方法。

　　同步代码一旦加锁后，那么会有一个临时的锁引用执行锁对象，和真实的引用无直接关联。在锁未释放之前，修改锁对象引用，不会影响同步代码的执行。

　　Java 虚拟机中的同步(Synchronization)基于进入和退出管程(Monitor)对象实现。同步方法并不是由 monitor enter 和 monitor exit 指令来实现同步的，而是由方法调用指令读取运行时常量池中方法的 ACC_SYNCHRONIZED 标志来隐式实现的。在 Java 虚拟机(HotSpot)中，monitor 是由 ObjectMonitor 实现的。

 /**

  * synchronized关键字

  * 锁对象。synchronized(this)和synchronized方法都是锁当前对象。

  */

 package concurrent.t01;

 import java.util.concurrent.TimeUnit;

 public class Test_01 {

     private int count = 0;

     private Object o = new Object();

     public void testSync1(){

         synchronized(o){

             System.out.println(Thread.currentThread().getName()

                     + " count = " + count++);

         }

     }

     public void testSync2(){

         synchronized(this){

             System.out.println(Thread.currentThread().getName()

                     + " count = " + count++);

         }

     }

     public synchronized void testSync3(){

         System.out.println(Thread.currentThread().getName()

                 + " count = " + count++);

         try {

             TimeUnit.SECONDS.sleep(3);

         } catch (InterruptedException e) {

             e.printStackTrace();

         }

     }

     public static void main(String[] args) {

         final Test_01 t = new Test_01();

         new Thread(new Runnable() {

             @Override

             public void run() {

                 t.testSync3();

             }

         }).start();

         new Thread(new Runnable() {

             @Override

             public void run() {

                 t.testSync3();

             }

         }).start();

     }

 }

synchronized关键字

1.2 volatile 关键字

　　变量的线程可见性。在 CPU 计算过程中，会将计算过程需要的数据加载到 CPU 计算缓存中，当 CPU 计算中断时，有可能刷新缓存，重新读取内存中的数据。在线程运行的过程中，如果某变量被其他线程修改，可能造成数据不一致的情况，从而导致结果错误。而 volatile修饰的变量是线程可见的，当 JVM 解释 volatile 修饰的变量时，会通知 CPU，在计算过程中，每次使用变量参与计算时，都会检查内存中的数据是否发生变化，而不是一直使用 CPU 缓存中的数据，可以保证计算结果的正确。volatile 只是通知底层计算时，CPU 检查内存数据，而不是让一个变量在多个线程中同步。

 /**

  * volatile关键字

  * volatile的可见性

  * 通知OS操作系统底层，在CPU计算过程中，都要检查内存中数据的有效性。保证最新的内存数据被使用。

  *

  */

 package concurrent.t01;

 import java.util.concurrent.TimeUnit;

 public class Test_09 {

     volatile boolean b = true;

     void m(){

         System.out.println("start");

         while(b){}

         System.out.println("end");

     }

     public static void main(String[] args) {

         final Test_09 t = new Test_09();

         new Thread(new Runnable() {

             @Override

             public void run() {

                 t.m();

             }

         }).start();

         try {

             TimeUnit.SECONDS.sleep(1);

         } catch (InterruptedException e) {

             // TODO Auto-generated catch block

             e.printStackTrace();

         }

         t.b = false;

     }

 }

volatile的可见性

 /**

  * volatile关键字

  * volatile的非原子性问题

  * volatile， 只能保证可见性，不能保证原子性。

  * 不是枷锁问题，只是内存数据可见。

  */

 package concurrent.t01;

 import java.util.ArrayList;

 import java.util.List;

 public class Test_10 {

     volatile int count = 0;

     /*synchronized*/ void m(){

         for(int i = 0; i < 10000; i++){

             count++;

         }

     }

     public static void main(String[] args) {

         final Test_10 t = new Test_10();

         List<Thread> threads = new ArrayList<>();

         for(int i = 0; i < 10; i++){

             threads.add(new Thread(new Runnable() {

                 @Override

                 public void run() {

                     t.m();

                 }

             }));

         }

         for(Thread thread : threads){

             thread.start();

         }

         for(Thread thread : threads){

             try {

                 thread.join();

             } catch (InterruptedException e) {

                 // TODO Auto-generated catch block

                 e.printStackTrace();

             }

         }

         System.out.println(t.count);

     }

 }

volatile的非原子性问题

1.3 wait&notify

　　当线程执行wait()方法时候，会释放当前的锁，然后让出CPU，进入等待状态。只有当 notify/notifyAll() 被执行时候，才会唤醒一个或多个正处于等待状态的线程，然后继续往下执行，直到执行完synchronized 代码块的代码或是中途遇到wait() ，再次释放锁。

　　由于 wait()、notify/notifyAll() 在synchronized 代码块执行，说明当前线程一定是获取了锁的。wait()、notify/notifyAll() 方法是Object的本地final方法，无法被重写。

 /**

  * 生产者消费者

  * wait&notify

  * wait/notify都是和while配合应用的。可以避免多线程并发判断逻辑失效问题。各位想想为什么不能用if

  */

 package concurrent.t04;

 import java.util.LinkedList;

 import java.util.concurrent.TimeUnit;

 public class TestContainer01<E> {

     private final LinkedList<E> list = new LinkedList<>();

     private final int MAX = 10;

     private int count = 0;

     public synchronized int getCount(){

         return count;

     }

     public synchronized void put(E e){

         while(list.size() == MAX){

             try {

                 this.wait();

             } catch (InterruptedException e1) {

                 e1.printStackTrace();

             }

         }

         list.add(e);

         count++;

         this.notifyAll();

     }

     public synchronized E get(){

         E e = null;

         while(list.size() == 0){

             try{

                 this.wait();

             } catch (InterruptedException e1) {

                 e1.printStackTrace();

             }

         }

         e = list.removeFirst();

         count--;

         this.notifyAll();

         return e;

     }

     public static void main(String[] args) {

         final TestContainer01<String> c = new TestContainer01<>();

         for(int i = 0; i < 10; i++){

             new Thread(new Runnable() {

                 @Override

                 public void run() {

                     for(int j = 0; j < 5; j++){

                         System.out.println(c.get());

                     }

                 }

             }, "consumer"+i).start();

         }

         try {

             TimeUnit.SECONDS.sleep(2);

         } catch (InterruptedException e) {

             // TODO Auto-generated catch block

             e.printStackTrace();

         }

         for(int i = 0; i < 2; i++){

             new Thread(new Runnable() {

                 @Override

                 public void run() {

                     for(int j = 0; j < 25; j++){

                         c.put("container value " + j);

                     }

                 }

             }, "producer"+i).start();

         }

     }

 }

wait&notify

1.4 AtomicXxx 类型

　　原子类型。

　　在 concurrent.atomic 包中定义了若干原子类型，这些类型中的每个方法都是保证了原子操作的。多线程并发访问原子类型对象中的方法，不会出现数据错误。在多线程开发中，如果某数据需要多个线程同时操作，且要求计算原子性，可以考虑使用原子类型对象。

AtomicBoolean，AtomicInteger，AtomicLong，AtomicReference
AtomicIntegerArray，AtomicLongArray
AtomicLongFieldUpdater，AtomicIntegerFieldUpdater，AtomicReferenceFieldUpdater
AtomicMarkableReference，AtomicStampedReference，AtomicReferenceArray

　　注意：原子类型中的方法是保证了原子操作，但多个方法之间是没有原子性的。即访问对2个或2个以上的atomic变量（或者对单个atomic变量进行2次或2次以上的操作），还是需要同步。

 /**

  * AtomicXxx

  * 同步类型

  * 原子操作类型。 其中的每个方法都是原子操作。可以保证线程安全。

  */

 package concurrent.t01;

 import java.util.ArrayList;

 import java.util.List;

 import java.util.concurrent.atomic.AtomicInteger;

 public class Test_11 {

     AtomicInteger count = new AtomicInteger(0);

     void m(){

         for(int i = 0; i < 10000; i++){

             /*if(count.get() < 1000)*/

                 count.incrementAndGet();

         }

     }

     public static void main(String[] args) {

         final Test_11 t = new Test_11();

         List<Thread> threads = new ArrayList<>();

         for(int i = 0; i < 10; i++){

             threads.add(new Thread(new Runnable() {

                 @Override

                 public void run() {

                     t.m();

                 }

             }));

         }

         for(Thread thread : threads){

             thread.start();

         }

         for(Thread thread : threads){

             try {

                 thread.join();

             } catch (InterruptedException e) {

                 // TODO Auto-generated catch block

                 e.printStackTrace();

             }

         }

         System.out.println(t.count.intValue());

     }

 }

AtomicXxx

1.5 CountDownLatch 门闩

　　门闩是 concurrent 包中定义的一个类型，是用于多线程通讯的一个辅助类型。门闩相当于在一个门上加多个锁，当线程调用 await 方法时，会检查门闩数量，如果门闩数量大于 0，线程会阻塞等待。当线程调用 countDown 时，会递减门闩的数量，当门闩数量为 0 时，await 阻塞线程可执行。

 /**

  * 门闩 - CountDownLatch

  * 可以和锁混合使用，或替代锁的功能。

  * 在门闩未完全开放之前等待。当门闩完全开放后执行。

  * 避免锁的效率低下问题。

  */

 package concurrent.t01;

 import java.util.concurrent.CountDownLatch;

 import java.util.concurrent.TimeUnit;

 public class Test_15 {

     CountDownLatch latch = new CountDownLatch(5);

     void m1(){

         try {

             latch.await();// 等待门闩开放。

         } catch (InterruptedException e) {

             e.printStackTrace();

         }

         System.out.println("m1() method");

     }

     void m2(){

         for(int i = 0; i < 10; i++){

             if(latch.getCount() != 0){

                 System.out.println("latch count : " + latch.getCount());

                 latch.countDown(); // 减门闩上的锁。

             }

             try {

                 TimeUnit.MILLISECONDS.sleep(500);

             } catch (InterruptedException e) {

                 // TODO Auto-generated catch block

                 e.printStackTrace();

             }

             System.out.println("m2() method : " + i);

         }

     }

     public static void main(String[] args) {

         final Test_15 t = new Test_15();

         new Thread(new Runnable() {

             @Override

             public void run() {

                 t.m1();

             }

         }).start();

         new Thread(new Runnable() {

             @Override

             public void run() {

                 t.m2();

             }

         }).start();

     }

 }

CountDownLatch

1.6 锁的重入

　　在 Java 中，同步锁是可以重入的。只有同一线程调用同步方法或执行同步代码块，对同一个对象加锁时才可重入。

　　当线程持有锁时，会在 monitor 的计数器中执行递增计算，若当前线程调用其他同步代码，且同步代码的锁对象相同时，monitor 中的计数器继续递增。每个同步代码执行结束，monitor 中的计数器都会递减，直至所有同步代码执行结束，monitor 中的计数器为 0 时，释放锁标记，_Owner 标记赋值为 null。

 /**

  * 锁可重入。 同一个线程，多次调用同步代码，锁定同一个锁对象，可重入。

  */

 package concurrent.t01;

 import java.util.concurrent.TimeUnit;

 public class Test_06 {

     synchronized void m1(){ // 锁this

         System.out.println("m1 start");

         try {

             TimeUnit.SECONDS.sleep(2);

         } catch (InterruptedException e) {

             e.printStackTrace();

         }

         m2();

         System.out.println("m1 end");

     }

     synchronized void m2(){ // 锁this

         System.out.println("m2 start");

         try {

             TimeUnit.SECONDS.sleep(1);

         } catch (InterruptedException e) {

             e.printStackTrace();

         }

         System.out.println("m2 end");

     }

     public static void main(String[] args) {

         new Test_06().m1();

     }

 }

锁可重入

1.7 ReentrantLock

　　重入锁，建议应用的同步方式。相对效率比 synchronized 高。量级较轻。使用重入锁，必须手工释放锁标记。一般都是在 finally 代码块中定义释放锁标记的 unlock 方法。

 /**

  * ReentrantLock

  *  重入锁

  */

 package concurrent.t03;

 import java.util.concurrent.TimeUnit;

 import java.util.concurrent.locks.Lock;

 import java.util.concurrent.locks.ReentrantLock;

 public class Test_01 {

     Lock lock = new ReentrantLock();

     void m1(){

         try{

             lock.lock(); // 加锁

             for(int i = 0; i < 10; i++){

                 TimeUnit.SECONDS.sleep(1);

                 System.out.println("m1() method " + i);

             }

         }catch(InterruptedException e){

             e.printStackTrace();

         }finally{

             lock.unlock(); // 解锁

         }

     }

     void m2(){

         lock.lock();

         System.out.println("m2() method");

         lock.unlock();

     }

     public static void main(String[] args) {

         final Test_01 t = new Test_01();

         new Thread(new Runnable() {

             @Override

             public void run() {

                 t.m1();

             }

         }).start();

         try {

             TimeUnit.SECONDS.sleep(1);

         } catch (InterruptedException e) {

             e.printStackTrace();

         }

         new Thread(new Runnable() {

             @Override

             public void run() {

                 t.m2();

             }

         }).start();

     }

 }

重入锁

 /**

  * 尝试锁

  */

 package concurrent.t03;

 import java.util.concurrent.TimeUnit;

 import java.util.concurrent.locks.Lock;

 import java.util.concurrent.locks.ReentrantLock;

 public class Test_02 {

     Lock lock = new ReentrantLock();

     void m1(){

         try{

             lock.lock();

             for(int i = 0; i < 10; i++){

                 TimeUnit.SECONDS.sleep(1);

                 System.out.println("m1() method " + i);

             }

         }catch(InterruptedException e){

             e.printStackTrace();

         }finally{

             lock.unlock();

         }

     }

     void m2(){

         boolean isLocked = false;

         try{

             // 尝试锁， 如果有锁，无法获取锁标记，返回false。

             // 如果获取锁标记，返回true

             // isLocked = lock.tryLock();

             // 阻塞尝试锁，阻塞参数代表的时长，尝试获取锁标记。

             // 如果超时，不等待。直接返回。

             isLocked = lock.tryLock(5, TimeUnit.SECONDS); 

             if(isLocked){

                 System.out.println("m2() method synchronized");

             }else{

                 System.out.println("m2() method unsynchronized");

             }

         }catch(Exception e){

             e.printStackTrace();

         }finally{

             if(isLocked){

                 // 尝试锁在解除锁标记的时候，一定要判断是否获取到锁标记。

                 // 如果当前线程没有获取到锁标记，会抛出异常。

                 lock.unlock();

             }

         }

     }

     public static void main(String[] args) {

         final Test_02 t = new Test_02();

         new Thread(new Runnable() {

             @Override

             public void run() {

                 t.m1();

             }

         }).start();

         try {

             TimeUnit.SECONDS.sleep(1);

         } catch (InterruptedException e) {

             // TODO Auto-generated catch block

             e.printStackTrace();

         }

         new Thread(new Runnable() {

             @Override

             public void run() {

                 t.m2();

             }

         }).start();

     }

 }

尝试锁

 /**

  * 可打断

  *

  * 阻塞状态： 包括普通阻塞，等待队列，锁池队列。

  * 普通阻塞： sleep(10000)， 可以被打断。调用thread.interrupt()方法，可以打断阻塞状态，抛出异常。

  * 等待队列： wait()方法被调用，也是一种阻塞状态，只能由notify唤醒。无法打断

  * 锁池队列： 无法获取锁标记。不是所有的锁池队列都可被打断。

  *  使用ReentrantLock的lock方法，获取锁标记的时候，如果需要阻塞等待锁标记，无法被打断。

  *  使用ReentrantLock的lockInterruptibly方法，获取锁标记的时候，如果需要阻塞等待，可以被打断。

  *

  */

 package concurrent.t03;

 import java.util.concurrent.TimeUnit;

 import java.util.concurrent.locks.Lock;

 import java.util.concurrent.locks.ReentrantLock;

 public class Test_03 {

     Lock lock = new ReentrantLock();

     void m1(){

         try{

             lock.lock();

             for(int i = 0; i < 5; i++){

                 TimeUnit.SECONDS.sleep(1);

                 System.out.println("m1() method " + i);

             }

         }catch(InterruptedException e){

             e.printStackTrace();

         }finally{

             lock.unlock();

         }

     }

     void m2(){

         try{

             lock.lockInterruptibly(); // 可尝试打断，阻塞等待锁。可以被其他的线程打断阻塞状态

             System.out.println("m2() method");

         }catch(InterruptedException e){

             System.out.println("m2() method interrupted");

         }finally{

             try{

                 lock.unlock();

             }catch(Exception e){

                 e.printStackTrace();

             }

         }

     }

     public static void main(String[] args) {

         final Test_03 t = new Test_03();

         new Thread(new Runnable() {

             @Override

             public void run() {

                 t.m1();

             }

         }).start();

         try {

             TimeUnit.SECONDS.sleep(1);

         } catch (InterruptedException e) {

             // TODO Auto-generated catch block

             e.printStackTrace();

         }

         Thread t2 = new Thread(new Runnable() {

             @Override

             public void run() {

                 t.m2();

             }

         });

         t2.start();

         try {

             TimeUnit.SECONDS.sleep(1);

         } catch (InterruptedException e) {

             // TODO Auto-generated catch block

             e.printStackTrace();

         }

         t2.interrupt();// 打断线程休眠。非正常结束阻塞状态的线程，都会抛出异常。

     }

 }

可打断

 /**

  * 公平锁

  */

 package concurrent.t03;

 import java.util.concurrent.locks.ReentrantLock;

 public class Test_04 {

     public static void main(String[] args) {

         TestReentrantlock t = new TestReentrantlock();

         //TestSync t = new TestSync();

         Thread t1 = new Thread(t);

         Thread t2 = new Thread(t);

         t1.start();

         t2.start();

     }

 }

 class TestReentrantlock extends Thread{

     // 定义一个公平锁

     private static ReentrantLock lock = new ReentrantLock(true);

     public void run(){

         for(int i = 0; i < 5; i++){

             lock.lock();

             try{

                 System.out.println(Thread.currentThread().getName() + " get lock");

             }finally{

                 lock.unlock();

             }

         }

     }

 }

 class TestSync extends Thread{

     public void run(){

         for(int i = 0; i < 5; i++){

             synchronized (this) {

                 System.out.println(Thread.currentThread().getName() + " get lock in TestSync");

             }

         }

     }

 }

公平锁

 /**

  * 生产者消费者

  * 重入锁&条件

  * 条件 - Condition， 为Lock增加条件。当条件满足时，做什么事情，如加锁或解锁。如等待或唤醒

  */

 package concurrent.t04;

 import java.io.IOException;

 import java.util.LinkedList;

 import java.util.concurrent.TimeUnit;

 import java.util.concurrent.locks.Condition;

 import java.util.concurrent.locks.Lock;

 import java.util.concurrent.locks.ReentrantLock;

 public class TestContainer02<E> {

     private final LinkedList<E> list = new LinkedList<>();

     private final int MAX = 10;

     private int count = 0;

     private Lock lock = new ReentrantLock();

     private Condition producer = lock.newCondition();

     private Condition consumer = lock.newCondition();

     public int getCount(){

         return count;

     }

     public void put(E e){

         lock.lock();

         try {

             while(list.size() == MAX){

                 System.out.println(Thread.currentThread().getName() + " 等待。。。");

                 // 进入等待队列。释放锁标记。

                 // 借助条件，进入的等待队列。

                 producer.await();

             }

             System.out.println(Thread.currentThread().getName() + " put 。。。");

             list.add(e);

             count++;

             // 借助条件，唤醒所有的消费者。

             consumer.signalAll();

         } catch (InterruptedException e1) {

             e1.printStackTrace();

         } finally {

             lock.unlock();

         }

     }

     public E get(){

         E e = null;

         lock.lock();

         try {

             while(list.size() == 0){

                 System.out.println(Thread.currentThread().getName() + " 等待。。。");

                 // 借助条件，消费者进入等待队列

                 consumer.await();

             }

             System.out.println(Thread.currentThread().getName() + " get 。。。");

             e = list.removeFirst();

             count--;

             // 借助条件，唤醒所有的生产者

             producer.signalAll();

         } catch (InterruptedException e1) {

             e1.printStackTrace();

         } finally {

             lock.unlock();

         }

         return e;

     }

     public static void main(String[] args) {

         final TestContainer02<String> c = new TestContainer02<>();

         for(int i = 0; i < 10; i++){

             new Thread(new Runnable() {

                 @Override

                 public void run() {

                     for(int j = 0; j < 5; j++){

                         System.out.println(c.get());

                     }

                 }

             }, "consumer"+i).start();

         }

         try {

             TimeUnit.SECONDS.sleep(2);

         } catch (InterruptedException e1) {

             e1.printStackTrace();

         }

         for(int i = 0; i < 2; i++){

             new Thread(new Runnable() {

                 @Override

                 public void run() {

                     for(int j = 0; j < 25; j++){

                         c.put("container value " + j);

                     }

                 }

             }, "producer"+i).start();

         }

     }

 }

重入锁&条件

1.8 ThreadLocal

　　ThreadLocal 提供了线程本地的实例。它与普通变量的区别在于，每个使用该变量的线程都会初始化一个完全独立的实例副本。ThreadLocal 变量通常被private static修饰。当一个线程结束时，它所使用的所有 ThreadLocal 相对的实例副本都可被回收。

　　ThreadLocal 适用于每个线程需要自己独立的实例且该实例需要在多个方法中被使用，也即变量在线程间隔离而在方法或类间共享的场景。每个 Thread 有自己的实例副本，且其它 Thread 不可访问，那就不存在多线程间共享的问题。

 /**

  * ThreadLocal

  * 就是一个Map。key - 》 Thread.getCurrentThread().  value - 》 线程需要保存的变量。

  * ThreadLocal.set(value) -> map.put(Thread.getCurrentThread(), value);

  * ThreadLocal.get() -> map.get(Thread.getCurrentThread());

  * 内存问题 ： 在并发量高的时候，可能有内存溢出。

  * 使用ThreadLocal的时候，一定注意回收资源问题，每个线程结束之前，将当前线程保存的线程变量一定要删除 。

  * ThreadLocal.remove();

  */

 package concurrent.t05;

 import java.util.concurrent.TimeUnit;

 public class Test_01 {

     volatile static String name = "zhangsan";

     static ThreadLocal<String> tl = new ThreadLocal<>();

     public static void main(String[] args) {

         new Thread(new Runnable() {

             @Override

             public void run() {

                 try {

                     TimeUnit.SECONDS.sleep(2);

                 } catch (InterruptedException e) {

                     e.printStackTrace();

                 }

                 System.out.println(name);

                 System.out.println(tl.get());

             }

         }).start();

         new Thread(new Runnable() {

             @Override

             public void run() {

                 try {

                     TimeUnit.SECONDS.sleep(1);

                 } catch (InterruptedException e) {

                     e.printStackTrace();

                 }

                 name = "lisi";

                 tl.set("wangwu");

             }

         }).start();

     }

 }

ThreadLocal

　　如果调用 ThreadLocal 的 set 方法将一个对象放入Thread中的成员变量threadLocals 中，那么这个对象是永远不会被回收的，因为这个对象永远都被Thread中的成员变量threadLocals引用着，可能会造成 OutOfMemoryError。需要调用 ThreadLocal 的 remove 方法将对象从thread中的成员变量threadLocals中删除掉。

二、同步容器

　　线程安全的容器对象： Vector, Hashtable。线程安全容器对象，都是使用 synchronized方法实现的。

　　concurrent 包中的同步容器，大多数是使用系统底层技术实现的线程安全。类似 native。Java8 中使用 CAS。

2.1 Map/Set

ConcurrentHashMap/ConcurrentHashSet

　　　　底层哈希实现的同步 Map(Set)。效率高，线程安全。使用系统底层技术实现线程安全。量级较 synchronized 低。key 和 value 不能为 null。

ConcurrentSkipListMap/ConcurrentSkipListSet

　　　　底层跳表（SkipList）实现的同步 Map(Set)。有序，效率比 ConcurrentHashMap 稍低。

 /**

  * 并发容器 - ConcurrentMap

  */

 package concurrent.t06;

 import java.util.HashMap;

 import java.util.Hashtable;

 import java.util.Map;

 import java.util.Random;

 import java.util.concurrent.ConcurrentHashMap;

 import java.util.concurrent.ConcurrentSkipListMap;

 import java.util.concurrent.CountDownLatch;

 public class Test_01_ConcurrentMap {

     public static void main(String[] args) {

         final Map<String, String> map = new Hashtable<>();

         // final Map<String, String> map = new ConcurrentHashMap<>();

         // final Map<String, String> map = new ConcurrentSkipListMap<>();

         final Random r = new Random();

         Thread[] array = new Thread[100];

         final CountDownLatch latch = new CountDownLatch(array.length);

         long begin = System.currentTimeMillis();

         for(int i = 0; i < array.length; i++){

             array[i] = new Thread(new Runnable() {

                 @Override

                 public void run() {

                     for(int j = 0; j < 10000; j++){

                         map.put("key"+r.nextInt(100000), "value"+r.nextInt(100000));

                     }

                     latch.countDown();

                 }

             });

         }

         for(Thread t : array){

             t.start();

         }

         try {

             latch.await();

         } catch (InterruptedException e) {

             e.printStackTrace();

         }

         long end = System.currentTimeMillis();

         System.out.println("执行时间为 ： " + (end-begin) + "毫秒！");

     }

 }

并发容器 - ConcurrentMap

2.2 List

CopyOnWriteArrayList

 /**

  * 并发容器 - CopyOnWriteList

  * 写时复制集合。写入效率低，读取效率高。每次写入数据，都会创建一个新的底层数组。

  */

 package concurrent.t06;

 import java.util.ArrayList;

 import java.util.List;

 import java.util.Random;

 import java.util.Vector;

 import java.util.concurrent.CopyOnWriteArrayList;

 import java.util.concurrent.CountDownLatch;

 public class Test_02_CopyOnWriteList {

     public static void main(String[] args) {

         // final List<String> list = new ArrayList<>();

         // final List<String> list = new Vector<>();

         final List<String> list = new CopyOnWriteArrayList<>();

         final Random r = new Random();

         Thread[] array = new Thread[100];

         final CountDownLatch latch = new CountDownLatch(array.length);

         long begin = System.currentTimeMillis();

         for(int i = 0; i < array.length; i++){

             array[i] = new Thread(new Runnable() {

                 @Override

                 public void run() {

                     for(int j = 0; j < 1000; j++){

                         list.add("value" + r.nextInt(100000));

                     }

                     latch.countDown();

                 }

             });

         }

         for(Thread t : array){

             t.start();

         }

         try {

             latch.await();

         } catch (InterruptedException e) {

             e.printStackTrace();

         }

         long end = System.currentTimeMillis();

         System.out.println("执行时间为 ： " + (end-begin) + "毫秒！");

         System.out.println("List.size() : " + list.size());

     }

 }

CopyOnWriteList

2.3 Queue

ConcurrentLinkedQueue　　基础链表同步队列。

 /**

  * 并发容器 - ConcurrentLinkedQueue

  *  队列 - 链表实现的。

  */

 package concurrent.t06;

 import java.util.Queue;

 import java.util.concurrent.ConcurrentLinkedQueue;

 public class Test_03_ConcurrentLinkedQueue {

     public static void main(String[] args) {

         Queue<String> queue = new ConcurrentLinkedQueue<>();

         for(int i = 0; i < 10; i++){

             queue.offer("value" + i);

         }

         System.out.println(queue);

         System.out.println(queue.size());

         // peek() -> 查看queue中的首数据

         System.out.println(queue.peek());

         System.out.println(queue.size());

         // poll() -> 获取queue中的首数据

         System.out.println(queue.poll());

         System.out.println(queue.size());

     }

 }

ConcurrentLinkedQueue

LinkedBlockingQueue　　阻塞队列，队列容量不足自动阻塞，队列容量为 0 自动阻塞。

 /**

  * 并发容器 - LinkedBlockingQueue

  *  阻塞容器。

  *  put & take - 自动阻塞。

  *  put自动阻塞， 队列容量满后，自动阻塞

  *  take自动阻塞方法， 队列容量为0后，自动阻塞。

  */

 package concurrent.t06;

 import java.util.Random;

 import java.util.concurrent.BlockingQueue;

 import java.util.concurrent.LinkedBlockingQueue;

 import java.util.concurrent.TimeUnit;

 public class Test_04_LinkedBlockingQueue {

     final BlockingQueue<String> queue = new LinkedBlockingQueue<>();

     final Random r = new Random();

     public static void main(String[] args) {

         final Test_04_LinkedBlockingQueue t = new Test_04_LinkedBlockingQueue();

         new Thread(new Runnable() {

             @Override

             public void run() {

                 while(true){

                     try {

                         t.queue.put("value"+t.r.nextInt(1000));

                         TimeUnit.SECONDS.sleep(1);

                     } catch (InterruptedException e) {

                         e.printStackTrace();

                     }

                 }

             }

         }, "producer").start();

         for(int i = 0; i < 3; i++){

             new Thread(new Runnable() {

                 @Override

                 public void run() {

                     while(true){

                         try {

                             System.out.println(Thread.currentThread().getName() +

                                     " - " + t.queue.take());

                         } catch (InterruptedException e) {

                             e.printStackTrace();

                         }

                     }

                 }

             }, "consumer"+i).start();

         }

     }

 }

LinkedBlockingQueue

ArrayBlockingQueue　　底层数组实现的有界队列

 /**

  * 并发容器 - ArrayBlockingQueue

  *  有界容器。

  * 当容量不足的时候，有阻塞能力。

  *add 方法在容量不足的时候，抛出异常。

  *put 方法在容量不足的时候，阻塞等待。

  *offer 方法，

  *单参数 offer 方法，不阻塞。容量不足的时候，返回 false。当前新增数据操作放弃。

  *三参数 offer 方法（offer(value,times,timeunit)），容量不足的时候，阻塞 times 时长（单

  *位为 timeunit），如果在阻塞时长内，有容量空闲，新增数据返回 true。如果阻塞时长范围

  *内，无容量空闲，放弃新增数据，返回 false。

  */

 package concurrent.t06;

 import java.util.concurrent.ArrayBlockingQueue;

 import java.util.concurrent.BlockingQueue;

 import java.util.concurrent.TimeUnit;

 public class Test_05_ArrayBlockingQueue {

     final BlockingQueue<String> queue = new ArrayBlockingQueue<>(3);

     public static void main(String[] args) {

         final Test_05_ArrayBlockingQueue t = new Test_05_ArrayBlockingQueue();

         for(int i = 0; i < 5; i++){

             // System.out.println("add method : " + t.queue.add("value"+i));

             /*try {

                 t.queue.put("put"+i);

             } catch (InterruptedException e) {

                 e.printStackTrace();

             }

             System.out.println("put method : " + i);*/

             // System.out.println("offer method : " + t.queue.offer("value"+i));

             try {

                 System.out.println("offer method : " +

                             t.queue.offer("value"+i, 1, TimeUnit.SECONDS));

             } catch (InterruptedException e) {

                 e.printStackTrace();

             }

         }

         System.out.println(t.queue);

     }

 }

ArrayBlockingQueue

DelayQueue　　延时队列。根据比较机制，实现自定义处理顺序的队列。常用于定时任务。

 /**

  * 并发容器 - DelayQueue

  *  *容器。

  */

 package concurrent.t06;

 import java.util.concurrent.BlockingQueue;

 import java.util.concurrent.DelayQueue;

 import java.util.concurrent.Delayed;

 import java.util.concurrent.TimeUnit;

 public class Test_06_DelayQueue {

     static BlockingQueue<MyTask_06> queue = new DelayQueue<>();

     public static void main(String[] args) throws InterruptedException {

         long value = System.currentTimeMillis();

         MyTask_06 task1 = new MyTask_06(value + 2000);

         MyTask_06 task2 = new MyTask_06(value + 1000);

         MyTask_06 task3 = new MyTask_06(value + 3000);

         MyTask_06 task4 = new MyTask_06(value + 2500);

         MyTask_06 task5 = new MyTask_06(value + 1500);

         queue.put(task1);

         queue.put(task2);

         queue.put(task3);

         queue.put(task4);

         queue.put(task5);

         System.out.println(queue);

         System.out.println(value);

         for(int i = 0; i < 5; i++){

             System.out.println(queue.take());

         }

     }

 }

 class MyTask_06 implements Delayed {

     private long compareValue;

     public MyTask_06(long compareValue){

         this.compareValue = compareValue;

     }

     /**

      * 比较大小。自动实现升序

      * 建议和getDelay方法配合完成。

      * 如果在DelayQueue是需要按时间完成的计划任务，必须配合getDelay方法完成。

      */

     @Override

     public int compareTo(Delayed o) {

         return (int)(this.getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS));

     }

     /**

      * 获取计划时长的方法。

      * 根据参数TimeUnit来决定，如何返回结果值。

      */

     @Override

     public long getDelay(TimeUnit unit) {

         return unit.convert(compareValue - System.currentTimeMillis(), TimeUnit.MILLISECONDS);

     }

     @Override

     public String toString(){

         return "Task compare value is : " + this.compareValue;

     }

 }

DelayQueue

LinkedTransferQueue　　转移队列，使用 transfer 方法，实现数据的即时处理。没有消费者，就阻塞。

 /**

  * 并发容器 - LinkedTransferQueue

  *  转移队列

  *  add - 队列会保存数据，不做阻塞等待。

  *  transfer - 是TransferQueue的特有方法。必须有消费者（take()方法的调用者）。

  *   如果没有任意线程消费数据，transfer方法阻塞。一般用于处理即时消息。

  */

 package concurrent.t06;

 import java.util.concurrent.LinkedTransferQueue;

 import java.util.concurrent.TimeUnit;

 import java.util.concurrent.TransferQueue;

 public class Test_07_TransferQueue {

     TransferQueue<String> queue = new LinkedTransferQueue<>();

     public static void main(String[] args) {

         final Test_07_TransferQueue t = new Test_07_TransferQueue();

         /*new Thread(new Runnable() {

             @Override

             public void run() {

                 try {

                     System.out.println(Thread.currentThread().getName() + " thread begin " );

                     System.out.println(Thread.currentThread().getName() + " - " + t.queue.take());

                 } catch (InterruptedException e) {

                     e.printStackTrace();

                 }

             }

         }, "output thread").start();

         try {

             TimeUnit.SECONDS.sleep(2);

         } catch (InterruptedException e) {

             e.printStackTrace();

         }

         try {

             t.queue.transfer("test string");

         } catch (InterruptedException e) {

             e.printStackTrace();

         }*/

         new Thread(new Runnable() {

             @Override

             public void run() {

                 try {

                     t.queue.transfer("test string");

                     // t.queue.add("test string");

                     System.out.println("add ok");

                 } catch (Exception e) {

                     e.printStackTrace();

                 }

             }

         }).start();

         try {

             TimeUnit.SECONDS.sleep(2);

         } catch (InterruptedException e) {

             e.printStackTrace();

         }

         new Thread(new Runnable() {

             @Override

             public void run() {

                 try {

                     System.out.println(Thread.currentThread().getName() + " thread begin " );

                     System.out.println(Thread.currentThread().getName() + " - " + t.queue.take());

                 } catch (InterruptedException e) {

                     e.printStackTrace();

                 }

             }

         }, "output thread").start();

     }

 }

LinkedTransferQueue

SynchronusQueue　　同步队列，是一个容量为 0 的队列。是一个特殊的 TransferQueue。必须现有消费线程等待，才能使用的队列。

 /**

  * 并发容器 - SynchronousQueue

  * 必须现有消费线程等待，才能使用的队列。

  * add 方法，无阻塞。若没有消费线程阻塞等待数据，则抛出异常。

  * put 方法，有阻塞。若没有消费线程阻塞等待数据，则阻塞。

  */

 package concurrent.t06;

 import java.util.concurrent.BlockingQueue;

 import java.util.concurrent.SynchronousQueue;

 import java.util.concurrent.TimeUnit;

 public class Test_08_SynchronusQueue {

     BlockingQueue<String> queue = new SynchronousQueue<>();

     public static void main(String[] args) {

         final Test_08_SynchronusQueue t = new Test_08_SynchronusQueue();

         new Thread(new Runnable() {

             @Override

             public void run() {

                 try {

                     System.out.println(Thread.currentThread().getName() + " thread begin " );

                     try {

                         TimeUnit.SECONDS.sleep(2);

                     } catch (InterruptedException e) {

                         e.printStackTrace();

                     }

                     System.out.println(Thread.currentThread().getName() + " - " + t.queue.take());

                 } catch (InterruptedException e) {

                     e.printStackTrace();

                 }

             }

         }, "output thread").start();

         /*try {

             TimeUnit.SECONDS.sleep(3);

         } catch (InterruptedException e) {

             e.printStackTrace();

         }*/

         // t.queue.add("test add");

         try {

             t.queue.put("test put");

         } catch (InterruptedException e) {

             e.printStackTrace();

         }

         System.out.println(Thread.currentThread().getName() + " queue size : " + t.queue.size());

     }

 }

SynchronousQueue

三、 ThreadPool&Executor

3.1 Executor

　　线程池*接口。定义方法，void execute(Runnable)。方法是用于处理任务的一个服务方法。调用者提供 Runnable 接口的实现，线程池通过线程执行这个 Runnable。服务方法无返回值的。是 Runnable 接口中的 run 方法无返回值。

　　常用方法 - void execute(Runnable)
　　作用是：启动线程任务的。

 /**

  * 线程池

  *     Executor - 线程池底层处理机制。

  * 在使用线程池的时候，底层如何调用线程中的逻辑。

  */

 package concurrent.t08;

 import java.util.concurrent.Executor;

 public class Test_01_MyExecutor implements Executor {

     public static void main(String[] args) {

         new Test_01_MyExecutor().execute(new Runnable() {

             @Override

             public void run() {

                 System.out.println(Thread.currentThread().getName() + " - test executor");

             }

         });

     }

     @Override

     public void execute(Runnable command) {

         new Thread(command).start();

     }

 }

线程池底层处理机制

3.2 ExecutorService

　　Executor 接口的子接口。提供了一个新的服务方法，submit。有返回值（Future 类型）。submit 方法提供了 overload 方法。其中有参数类型为 Runnable 的，不需要提供返回值的；有参数类型为 Callable，可以提供线程执行后的返回值。

Future，是 submit 方法的返回值。代表未来，也就是线程执行结束后的一种结果。如返回值。
常见方法 - void execute(Runnable)， Future submit(Callable)， Future submit(Runnable)
线程池状态： Running， ShuttingDown， Termitnaed

ShuttingDown - 线程池正在关闭过程中。优雅关闭。一旦进入这个状态，线程池不再接收新的任务，处理所有已接收的任务，处理完毕后，关闭线程池。
Terminated - 线程池已经关闭。

3.3 Future

　　未来结果，代表线程任务执行结束后的结果。获取线程执行结果的方式是通过 get 方法获取的。get 无参，阻塞等待线程执行结束，并得到结果。get 有参，阻塞固定时长，等待
线程执行结束后的结果，如果在阻塞时长范围内，线程未执行结束，抛出异常。
　　常用方法： T get() T get(long, TimeUnit)

 /**

  * 线程池

  * 固定容量线程池

  */

 package concurrent.t08;

 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;

 import java.util.concurrent.FutureTask;

 import java.util.concurrent.TimeUnit;

 public class Test_03_Future {

     public static void main(String[] args) throws InterruptedException, ExecutionException {

         /*FutureTask<String> task = new FutureTask<>(new Callable<String>() {

             @Override

             public String call() throws Exception {

                 return "first future task";

             }

         });

         new Thread(task).start();

         System.out.println(task.get());*/

         ExecutorService service = Executors.newFixedThreadPool(1);

         Future<String> future = service.submit(new Callable<String>() {

             @Override

             public String call() {

                 try {

                     TimeUnit.MILLISECONDS.sleep(500);

                 } catch (InterruptedException e) {

                     e.printStackTrace();

                 }

                 System.out.println("aaa");

                 return Thread.currentThread().getName() + " - test executor";

             }

         });

         System.out.println(future);

         System.out.println(future.isDone()); // 查看线程是否结束， 任务是否完成。 call方法是否执行结束

         System.out.println(future.get()); // 获取call方法的返回值。

         System.out.println(future.isDone());

     }

 }

Future

3.4 Callable

　　可执行接口。类似 Runnable 接口。也是可以启动一个线程的接口。其中定义的方法是call。call 方法的作用和 Runnable 中的 run 方法完全一致。call 方法有返回值。
　　接口方法： Object call();相当于 Runnable 接口中的 run 方法。区别为此方法有返回值。不能抛出已检查异常。
　　和 Runnable 接口的选择 - 需要返回值或需要抛出异常时，使用 Callable，其他情况可任意选择。

3.5 Executors

　　工具类型。为 Executor 线程池提供工具方法。可以快速的提供若干种线程池。如：固定容量的，无限容量的，容量为 1 等各种线程池。
　　线程池是一个进程级的重量级资源。默认的生命周期和 JVM 一致。当开启线程池后，直到 JVM 关闭为止，是线程池的默认生命周期。如果手工调用 shutdown 方法，那么线程池执行所有的任务后，自动关闭。
　　开始 - 创建线程池。
　　结束 - JVM 关闭或调用 shutdown 并处理完所有的任务。
　　类似 Arrays，Collections 等工具类型的功用。

3.6 FixedThreadPool

　　容量固定的线程池。活动状态和线程池容量是有上限的线程池。所有的线程池中，都有一个任务队列。使用的是 BlockingQueue<Runnable>作为任务的载体。当任务数量大于线程池容量的时候，没有运行的任务保存在任务队列中，当线程有空闲的，自动从队列中取出任务执行。
　　使用场景：大多数情况下，使用的线程池，首选推荐 FixedThreadPool。OS 系统和硬件是有线程支持上限。不能随意的无限制提供线程池。

　　线程池默认的容量上限是 Integer.MAX_VALUE。
　　常见的线程池容量： PC - 200。服务器 - 1000~10000
　　queued tasks - 任务队列
　　completed tasks - 结束任务队列

 /**

  * 线程池

  * 固定容量线程池

  * FixedThreadPool - 固定容量线程池。创建线程池的时候，容量固定。

  *  构造的时候，提供线程池最大容量

  * new xxxxx ->

  * ExecutorService - 线程池服务类型。所有的线程池类型都实现这个接口。

  *  实现这个接口，代表可以提供线程池能力。

  *  shutdown - 优雅关闭。 不是强行关闭线程池，回收线程池中的资源。而是不再处理新的任务，将已接收的任务处理完毕后

  *      再关闭。

  * Executors - Executor的工具类。类似Collection和Collections的关系。

  *  可以更简单的创建若干种线程池。

  */

 package concurrent.t08;

 import java.util.concurrent.ExecutorService;

 import java.util.concurrent.Executors;

 import java.util.concurrent.TimeUnit;

 public class Test_02_FixedThreadPool {

     public static void main(String[] args) {

         ExecutorService service =

                 Executors.newFixedThreadPool(5);

         for(int i = 0; i < 6; i++){

             service.execute(new Runnable() {

                 @Override

                 public void run() {

                     try {

                         TimeUnit.MILLISECONDS.sleep(500);

                     } catch (InterruptedException e) {

                         e.printStackTrace();

                     }

                     System.out.println(Thread.currentThread().getName() + " - test executor");

                 }

             });

         }

         System.out.println(service);

         service.shutdown();

         // 是否已经结束， 相当于回收了资源。

         System.out.println(service.isTerminated());

         // 是否已经关闭， 是否调用过shutdown方法

         System.out.println(service.isShutdown());

         System.out.println(service);

         try {

             TimeUnit.SECONDS.sleep(2);

         } catch (InterruptedException e) {

             e.printStackTrace();

         }

         // service.shutdown();

         System.out.println(service.isTerminated());

         System.out.println(service.isShutdown());

         System.out.println(service);

     }

 }

FixedThreadPool

3.7 CachedThreadPool

　　缓存的线程池。容量不限（Integer.MAX_VALUE）。自动扩容。容量管理策略：如果线程池中的线程数量不满足任务执行，创建新的线程。每次有新任务无法即时处理的时候，都会创建新的线程。当线程池中的线程空闲时长达到一定的临界值（默认 60 秒），自动释放线程。默认线程空闲 60 秒，自动销毁。
　　应用场景：内部应用或测试应用。内部应用，有条件的内部数据瞬间处理时应用，如：
　　电信平台夜间执行数据整理（有把握在短时间内处理完所有工作，且对硬件和软件有足够的信心）。测试应用，在测试的时候，尝试得到硬件或软件的最高负载量，用于提供FixedThreadPool 容量的指导。

 /**

  * 线程池

  * 无容量限制的线程池（最大容量默认为Integer.MAX_VALUE）

  */

 package concurrent.t08;

 import java.util.concurrent.ExecutorService;

 import java.util.concurrent.Executors;

 import java.util.concurrent.TimeUnit;

 public class Test_05_CachedThreadPool {

     public static void main(String[] args) {

         ExecutorService service = Executors.newCachedThreadPool();

         System.out.println(service);

         for(int i = 0; i < 5; i++){

             service.execute(new Runnable() {

                 @Override

                 public void run() {

                     try {

                         TimeUnit.MILLISECONDS.sleep(500);

                     } catch (InterruptedException e) {

                         e.printStackTrace();

                     }

                     System.out.println(Thread.currentThread().getName() + " - test executor");

                 }

             });

         }

         System.out.println(service);

         try {

             TimeUnit.SECONDS.sleep(65);

         } catch (InterruptedException e) {

             e.printStackTrace();

         }

         System.out.println(service);

     }

 }

CachedThreadPool

3.8 ScheduledThreadPool

　　计划任务线程池。可以根据计划自动执行任务的线程池。
　　scheduleAtFixedRate(Runnable, start_limit, limit, timeunit)
　　runnable - 要执行的任务。
　　start_limit - 第一次任务执行的间隔。
　　limit - 多次任务执行的间隔。
　　timeunit - 多次任务执行间隔的时间单位。
　　使用场景：计划任务时选用（DelaydQueue），如：电信行业中的数据整理，没分钟整理，没消失整理，每天整理等。

 /**

  * 线程池

  * 计划任务线程池。

  */

 package concurrent.t08;

 import java.util.concurrent.Executors;

 import java.util.concurrent.ScheduledExecutorService;

 import java.util.concurrent.TimeUnit;

 public class Test_07_ScheduledThreadPool {

     public static void main(String[] args) {

         ScheduledExecutorService service = Executors.newScheduledThreadPool(3);

         System.out.println(service);

         // 定时完成任务。 scheduleAtFixedRate(Runnable, start_limit, limit, timeunit)

         // runnable - 要执行的任务。

         service.scheduleAtFixedRate(new Runnable() {

             @Override

             public void run() {

                 try {

                     TimeUnit.MILLISECONDS.sleep(500);

                 } catch (InterruptedException e) {

                     e.printStackTrace();

                 }

                 System.out.println(Thread.currentThread().getName());

             }

         }, 0, 300, TimeUnit.MILLISECONDS);

     }

 }

ScheduledThreadPool

3.9 SingleThreadExceutor

　　单一容量的线程池。使用场景：保证任务顺序时使用。如：游戏大厅中的公共频道聊天。秒杀。

 /**

  * 线程池

  * 容量为1的线程池。 顺序执行。

  */

 package concurrent.t08;

 import java.util.concurrent.ExecutorService;

 import java.util.concurrent.Executors;

 import java.util.concurrent.TimeUnit;

 public class Test_06_SingleThreadExecutor {

     public static void main(String[] args) {

         ExecutorService service = Executors.newSingleThreadExecutor();

         System.out.println(service);

         for(int i = 0; i < 5; i++){

             service.execute(new Runnable() {

                 @Override

                 public void run() {

                     try {

                         TimeUnit.MILLISECONDS.sleep(500);

                     } catch (InterruptedException e) {

                         e.printStackTrace();

                     }

                     System.out.println(Thread.currentThread().getName() + " - test executor");

                 }

             });

         }

     }

 }

SingleThreadExecutor

3.10 ForkJoinPool

　　分支合并线程池（mapduce 类似的设计思想）。适合用于处理复杂任务。

　　初始化线程容量与 CPU 核心数相关。
　　线程池中运行的内容必须是 ForkJoinTask 的子类型（RecursiveTask,RecursiveAction）。ForkJoinPool - 分支合并线程池。可以递归完成复杂任务。要求可分支合并的任务必须是 ForkJoinTask 类型的子类型。其中提供了分支和合并的能力。ForkJoinTask 类型提供了两个抽象子类型，RecursiveTask 有返回结果的分支合并任务,RecursiveAction 无返回结果的分支合并任务。（Callable/Runnable）compute 方法：就是任务的执行逻辑。
　　ForkJoinPool 没有所谓的容量。默认都是 1 个线程。根据任务自动的分支新的子线程。当子线程任务结束后，自动合并。所谓自动是根据 fork 和 join 两个方法实现的。
　　应用：主要是做科学计算或天文计算的。数据分析的。

 /**

  * 线程池

  * 分支合并线程池。

  */

 package concurrent.t08;

 import java.io.IOException;

 import java.util.Random;

 import java.util.concurrent.ExecutionException;

 import java.util.concurrent.ForkJoinPool;

 import java.util.concurrent.Future;

 import java.util.concurrent.RecursiveTask;

 public class Test_08_ForkJoinPool {

     final static int[] numbers = new int[1000000];

     final static int MAX_SIZE = 50000;

     final static Random r = new Random();

     static{

         for(int i = 0; i < numbers.length; i++){

             numbers[i] = r.nextInt(1000);

         }

     }

     static class AddTask extends RecursiveTask<Long>{ // RecursiveAction

         int begin, end;

         public AddTask(int begin, int end){

             this.begin = begin;

             this.end = end;

         }

         //

         protected Long compute(){

             if((end - begin) < MAX_SIZE){

                 long sum = 0L;

                 for(int i = begin; i < end; i++){

                     sum += numbers[i];

                 }

                 // System.out.println("form " + begin + " to " + end + " sum is : " + sum);

                 return sum;

             }else{

                 int middle = begin + (end - begin)/2;

                 AddTask task1 = new AddTask(begin, middle);

                 AddTask task2 = new AddTask(middle, end);

                 task1.fork();// 就是用于开启新的任务的。 就是分支工作的。 就是开启一个新的线程任务。

                 task2.fork();

                 // join - 合并。将任务的结果获取。 这是一个阻塞方法。一定会得到结果数据。

                 return task1.join() + task2.join();

             }

         }

     }

     public static void main(String[] args) throws InterruptedException, ExecutionException, IOException {

         long result = 0L;

         for(int i = 0; i < numbers.length; i++){

             result += numbers[i];

         }

         System.out.println(result);

         ForkJoinPool pool = new ForkJoinPool();

         AddTask task = new AddTask(0, numbers.length);

         Future<Long> future = pool.submit(task);

         System.out.println(future.get());

     }

 }

ForkJoinPool

3.11 ThreadPoolExecutor

　线程池底层实现。除 ForkJoinPool 外，其他常用线程池底层都是使用 ThreadPoolExecutor实现的。

　　public ThreadPoolExecutor

　　(int corePoolSize, // 核心容量，创建线程池的时候，默认有多少线程。也是线程池保持的最少线程数

　　int maximumPoolSize, // 最大容量，线程池最多有多少线程

　　long keepAliveTime, // 生命周期，0 为永久。当线程空闲多久后，自动回收。

　　TimeUnit unit, // 生命周期单位，为生命周期提供单位，如：秒，毫秒

　　BlockingQueue<Runnable> workQueue // 任务队列，阻塞队列。注意，泛型必须是

　　Runnable

　　);

//使用场景： 默认提供的线程池不满足条件时使用。如：初始线程数据 4，最大线程数200，线程空闲周期 30 秒。

 /**

  * 线程池

  * 固定容量线程池

  */

 package concurrent.t08;

 import java.util.ArrayList;

 import java.util.concurrent.ExecutorService;

 import java.util.concurrent.LinkedBlockingQueue;

 import java.util.concurrent.ThreadPoolExecutor;

 import java.util.concurrent.TimeUnit;

 public class Test_09_ThreadPoolExecutor {

     public static void main(String[] args) {

         // 模拟fixedThreadPool， 核心线程5个，最大容量5个，线程的生命周期无限。

         ExecutorService service =

                 new ThreadPoolExecutor(5, 5, 0L, TimeUnit.MILLISECONDS,

                         new LinkedBlockingQueue<Runnable>());

         for(int i = 0; i < 6; i++){

             service.execute(new Runnable() {

                 @Override

                 public void run() {

                     try {

                         TimeUnit.MILLISECONDS.sleep(500);

                     } catch (InterruptedException e) {

                         e.printStackTrace();

                     }

                     System.out.println(Thread.currentThread().getName() + " - test executor");

                 }

             });

         }

         System.out.println(service);

         service.shutdown();

         System.out.println(service.isTerminated());

         System.out.println(service.isShutdown());

         System.out.println(service);

         try {

             TimeUnit.SECONDS.sleep(2);

         } catch (InterruptedException e) {

             e.printStackTrace();

         }

         service.shutdown();

         System.out.println(service.isTerminated());

         System.out.println(service.isShutdown());

         System.out.println(service);

     }

 }

ThreadPoolExecutor

码农公寓

相关文章