java面试宝典beta5.0

* Subclasses of Thread should override this method. * * @see #start() * @see #stop() * @see #Thread(ThreadGroup, Runnable, String) */ @Override public void run() { if (target != null) { target.run(); } 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 106 } ( 2 ) 定实现时器 Timer 和 TimerTask Timer 在实际开发中应用场景不多，一般来说都会用其他第三方库来实现。但有时会在一些面试题中出现。 下面我们就针对一道面试题来使用 Timer 定时类。 1. 请模拟写出双重定时器（面试题） 要求：使用定时器,间隔 4 秒执行一次，再间隔 2 秒执行一次，以此类推执行。 class TimerTastCus extends TimerTask{ @Override public void run() { count = (count +1)%2; System.err.println("Boob boom "); new Timer().schedule(new TimerTastCus(), 2000+2000*count); } } Timer timer = new Timer(); timer.schedule(new TimerTastCus(), 2000+2000*count); while (true) { System.out.println(new Date().getSeconds()); try { Thread.sleep(1000); } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } } //PS：下面的代码中的 count 变量中 //此参数要使用在你匿名内部类中，使用 final 修饰就无法对其值进行修改， //只能改为静态变量 private static volatile int count = 0; ( 3 ) 线程互斥与同步 在引入多线程后，由于线程执行的异步性，会给系统造成混乱，特别是在急用临界资源时，如多个线程急用同一台 打印机，会使打印结果交织在一起，难于区分。当多个线程急用共享变量，表格，链表时，可能会导致数据处理出错， 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 107 因此线程同步的主要任务是使并发执行的各线程之间能够有效的共享资源和相互合作，从而使程序的执行具有可再现性。 当线程并发执行时，由于资源共享和线程协作，使用线程之间会存在以下两种制约关系。 1. 间接相互制约。一个系统中的多个线程必然要共享某种系统资源，如共享 CPU，共享 I/O 设备，所谓间接相 互制约即源于这种资源共享，打印机就是最好的例子，线程 A 在使用打印机时，其它线程都要等待。 2. 直接相互制约。这种制约主要是因为线程之间的合作，如有线程 A 将计算结果提供给线程 B 作进一步处理， 那么线程 B 在线程 A 将数据送达之前都将处于阻塞状态。 间接相互制约可以称为互斥，直接相互制约可以称为同步，对于互斥可以这样理解，线程 A 和线程 B 互斥访问某 个资源则它们之间就会产个顺序问题——要么线程 A 等待线程 B 操作完毕，要么线程 B 等待线程操作完毕，这其实就 是线程的同步了。因此同步包括互斥，互斥其实是一种特殊的同步。 下面我们通过一道面试题来体会线程的交互。 要求：子线程运行执行 10 次后，主线程再运行 5 次。这样交替执行三遍 public static void main(String[] args) { final Bussiness bussiness = new Bussiness(); //子线程 new Thread(new Runnable() { @Override public void run() { for (int i = 0; i < 3; i++) { bussiness.subMethod(); } } }).start(); //主线程 for (int i = 0; i < 3; i++) { bussiness.mainMethod(); } } } class Bussiness { 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 108 private boolean subFlag = true; public synchronized void mainMethod() { while (subFlag) { try { wait(); } catch (InterruptedException e) { e.printStackTrace(); } } for (int i = 0; i < 5; i++) { System.out.println(Thread.currentThread().getName() + " : main thread running loop count -- " + i); try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } } subFlag = true; notify(); } public synchronized void subMethod() { while (!subFlag) { try { wait(); } catch (InterruptedException e) { e.printStackTrace(); } } for (int i = 0; i < 10; i++) { System.err.println(Thread.currentThread().getName() + " : sub thread running loop count -- " + i); try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } } subFlag = false; 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 109 notify(); } } ( 4 ) 线程局部变量 ThreadLocal  ThreadLocal 的作用和目的：用于实现线程内的数据共享，即对于相同的程序代码，多个模块在同一个 线程中运行时要共享一份数据，而在另外线程中运行时又共享另外一份数据。  每个线程调用全局 ThreadLocal 对象的 set 方法，在 set 方法中，首先根据当前线程获取当前线程的 ThreadLocalMap 对象，然后往这个 map 中插入一条记录，key 其实是 ThreadLocal 对象，value 是各自的 set 方法传进去的值。也就是每个线程其实都有一份自己独享的 ThreadLocalMap 对象，该对象的 Key 是 ThreadLocal 对象，值是用户设置的具体值。在线程结束时可以调用 ThreadLocal.remove()方法，这样会更快释放内存，不调 用也可以，因为线程结束后也可以自动释放相关的 ThreadLocal 变量。  ThreadLocal 的应用场景： ➢ 订单处理包含一系列操作：减少库存量、增加一条流水台账、修改总账，这几个操作要在同一个 事务中完成，通常也即同一个线程中进行处理，如果累加公司应收款的操作失败了，则应该把前面 的操作回滚，否则，提交所有操作，这要求这些操作使用相同的数据库连接对象，而这些操作的代码 分别位于不同的模块类中。 ➢ 银行转账包含一系列操作： 把转出帐户的余额减少，把转入帐户的余额增加，这两个操作要在 同一个事务中完成，它们必须使用相同的数据库连接对象，转入和转出操作的代码分别是两个不同 的帐户对象的方法。 ➢ 例如 Strut2 的 ActionContext，同一段代码被不同的线程调用运行时，该代码操作的数据是每 个线程各自的状态和数据，对于不同的线程来说，getContext 方法拿到的对象都不相同，对同一个 线程来说，不管调用 getContext 方法多少次和在哪个模块中 getContext 方法，拿到的都是同一 个。 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 110 1. ThreadLocal 的使用方式 (1) 在关联数据类中创建 private static ThreadLocal 在下面的类中，私有静态 ThreadLocal 实例（serialNum）为调用该类的静态 SerialNum.get() 方法的每个 线程维护了一个“序列号”，该方法将返回当前线程的序列号。（线程的序列号是在第一次调用 SerialNum.get() 时 分配的，并在后续调用中不会更改。） public class SerialNum { // The next serial number to be assigned private static int nextSerialNum = 0; private static ThreadLocal serialNum = new ThreadLocal() { protected synchronized Object initialValue() { return new Integer(nextSerialNum++); } }; public static int get() { return ((Integer) (serialNum.get())).intValue(); } } 另一个例子，也是私有静态 ThreadLocal 实例： public class ThreadContext { private String userId; private Long transactionId; private static ThreadLocal threadLocal = new ThreadLocal(){ @Override protected ThreadContext initialValue() { return new ThreadContext(); } }; public static ThreadContext get() { return threadLocal.get(); } public String getUserId() { 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 111 return userId; } public void setUserId(String userId) { this.userId = userId; } public Long getTransactionId() { return transactionId; } public void setTransactionId(Long transactionId) { this.transactionId = transactionId; } } 补充：在 JDK 的 API 对 ThreadLocal 私有化的说明。并举例‘线程唯一标识符’ UniqueThreadIdGenerator ，大家学习是可以结合官方 API 来学习。 2. 在 Util 类中创建 ThreadLocal 这是上面用法的扩展，即把 ThreadLocal 的创建放到工具类中。 public class HibernateUtil { private static Log log = LogFactory.getLog(HibernateUtil.class); private static final SessionFactory sessionFactory; //定义 SessionFactory static { try { // 通过默认配置文件 hibernate.cfg.xml 创建 SessionFactory sessionFactory = new Configuration().configure().buildSessionFactory(); } catch (Throwable ex) { log.error("初始化 SessionFactory 失败！", ex); throw new ExceptionInInitializerError(ex); } } //创建线程局部变量 session，用来保存 Hibernate 的 Session public static final ThreadLocal session = new ThreadLocal(); /** * 获取当前线程中的 Session * @return Session * @throws HibernateException */ public static Session currentSession() throws HibernateException { 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 112 Session s = (Session) session.get(); // 如果 Session 还没有打开，则新开一个 Session if (s == null) { s = sessionFactory.openSession(); session.set(s); //将新开的 Session 保存到线程局部变量中 } return s; } public static void closeSession() throws HibernateException { //获取线程局部变量，并强制转换为 Session 类型 Session s = (Session) session.get(); session.set(null); if (s != null) s.close(); } } 3. 在 Runnable 中创建 ThreadLocal 在线程类内部创建 ThreadLocal，基本步骤如下： ①、在多线程的类（如 ThreadDemo 类）中，创建一个 ThreadLocal 对象 threadXxx，用来保存线程间 需要隔离处理的对象 xxx。 ②、在 ThreadDemo 类中，创建一个获取要隔离访问的数据的方法 getXxx()，在方法中判断，若 ThreadLocal 对象为 null 时候，应该 new()一个隔离访问类型的对象，并强制转换为要应用的类型 ③、在 ThreadDemo 类的 run()方法中，通过调用 getXxx()方法获取要操作的数据，这样可以保证每个线 程对应一个数据对象，在任何时刻都操作的是这个对象。 public class ThreadLocalTest implements Runnable{ ThreadLocal studenThreadLocal = new ThreadLocal(); @Override public void run() { String currentThreadName = Thread.currentThread().getName(); System.out.println(currentThreadName + " is running..."); 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 113 Random random = new Random(); int age = random.nextInt(100); System.out.println(currentThreadName + " is set age: " + age); Studen studen = getStudent(); //通过这个方法，为每个线程都独立的 new 一个 student 对象，每个线程的的 student 对象都可以设置不同的值 studen.setAge(age); System.out.println(currentThreadName + " is first get age: " + studen.getAge()); try { Thread.sleep(500); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println( currentThreadName + " is second get age: " + studen.getAge()); } private Studen getStudent() { Studen studen = studenThreadLocal.get(); if (null == studen) { studen = new Studen(); studenThreadLocal.set(studen); } return studen; } public static void main(String[] args) { ThreadLocalTest t = new ThreadLocalTest(); Thread t1 = new Thread(t,"Thread A"); Thread t2 = new Thread(t,"Thread B"); t1.start(); t2.start(); } } class Studen{ int age; public int getAge() { return age; } public void setAge(int age) { this.age = age; } 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 114 } ( 5 ) 多线程共享数据 在 Java 传统线程机制中的共享数据方式，大致可以简单分两种情况： ➢ 多个线程行为一致，共同操作一个数据源。也就是每个线程执行的代码相同，可以使用同一个 Runnable 对 象，这个 Runnable 对象中有那个共享数据，例如，卖票系统就可以这么做。 ➢ 多个线程行为不一致，共同操作一个数据源。也就是每个线程执行的代码不同，这时候需要用不同的 Runnable 对象。例如，银行存取款。 下面我们通过两个示例代码来分别说明这两种方式。 1. 多个线程行为一致共同操作一个数据 如果每个线程执行的代码相同，可以使用同一个 Runnable 对象，这个 Runnable 对象中有那个共享数据，例如， 买票系统就可以这么做。 /** *共享数据类 **/ class ShareData{ private int num = 10 ; public synchronized void inc() { num++; System.out.println(Thread.currentThread().getName()+": invoke inc method num =" + num); try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } } } /** *多线程类 **/ class RunnableCusToInc implements Runnable{ private ShareData shareData; 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 115 public RunnableCusToInc(ShareData data) { this.shareData = data; } @Override public void run() { for (int i = 0; i < 5; i++) { shareData.inc(); } } } /** *测试方法 **/ public static void main(String[] args) { ShareData shareData = new ShareData(); for (int i = 0; i < 4; i++) { new Thread(new RunnableCusToInc(shareData),"Thread "+ i).start(); } } } 2. 多个线程行为不一致共同操作一个数据 如果每个线程执行的代码不同，这时候需要用不同的 Runnable 对象，有如下两种方式来实现这些 Runnable 对 象之间的数据共享： 1) 将共享数据封装在另外一个对象中，然后将这个对象逐一传递给各个 Runnable 对象。每个线程对共享 数据的操作方法也分配到那个对象身上去完成，这样容易实现针对该数据进行的各个操作的互斥和通信。 public static void main(String[] args) { ShareData shareData = new ShareData(); for (int i = 0; i < 4; i++) { if(i%2 == 0){ new Thread(new RunnableCusToInc(shareData),"Thread "+ i).start(); }else{ new Thread(new RunnableCusToDec(shareData),"Thread "+ i).start(); } 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 116 } } //封装共享数据类 class RunnableCusToInc implements Runnable{ //封装共享数据 private ShareData shareData; public RunnableCusToInc(ShareData data) { this.shareData = data; } @Override public void run() { for (int i = 0; i < 5; i++) { shareData.inc(); } } } //封装共享数据类 class RunnableCusToDec implements Runnable{ //封装共享数据 private ShareData shareData; public RunnableCusToDec(ShareData data) { this.shareData = data; } @Override public void run() { for (int i = 0; i < 5; i++) { shareData.dec(); } } } /** *共享数据类 **/ class ShareData{ private int num = 10 ; public synchronized void inc() { num++; System.out.println(Thread.currentThread().getName()+": invoke inc method num =" + num); 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 117 try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } } } 2) 将这些 Runnable 对象作为某一个类中的内部类，共享数据作为这个外部类中的成员变量，每个线程对 共享数据的操作方法也分配给外部类，以便实现对共享数据进行的各个操作的互斥和通信，作为内部类的各个 Runnable 对象调用外部类的这些方法。 public static void main(String[] args) { //公共数据 final ShareData shareData = new ShareData(); for (int i = 0; i < 4; i++) { if(i%2 == 0){ new Thread(new Runnable() { @Override public void run() { for (int i = 0; i < 5; i++) { shareData.inc(); } } },"Thread "+ i).start(); }else{ new Thread(new Runnable() { @Override public void run() { for (int i = 0; i < 5; i++) { shareData.dec(); } } },"Thread "+ i).start(); } } } 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 118 class ShareData{ private int num = 10 ; public synchronized void inc() { num++; System.out.println(Thread.currentThread().getName()+": invoke inc method num =" + num); try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } } public synchronized void dec() { num--; System.err.println(Thread.currentThread().getName()+": invoke dec method num =" + num); try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } } } 补充：上面两种方式的组合：将共享数据封装在另外一个对象中，每个线程对共享数据的操作方法也分配到 那个对象身上去完成，对象作为这个外部类中的成员变量或方法中的局部变量，每个线程的 Runnable 对象 作为外部类中的成员内部类或局部内部类。 总之，要同步互斥的几段代码最好是分别放在几个独立的方法中，这些方法再放在同一个类中，这样比较容 易实现它们之间的同步互斥和通信。 （二）多线程基础知识--线程并发库（2017-12-11-wl） Java 5 添加了一个新的包到 Java 平台，java.util.concurrent 包。这个包包含有一系列能够让 Java 的并发 编程变得更加简单轻松的类。在这个包被添加以前，你需要自己去动手实现自己的相关工具类。下面带你认识下 java.util.concurrent 包里的这些类，然后你可以尝试着如何在项目中使用它们。本文中将使用 Java 6 版本，我不确 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 119 定这和 Java 5 版本里的是否有一些差异。我不会去解释关于 Java 并发的核心问题 – 其背后的原理，也就是说， 如果你对那些东西感兴趣，参考《Java 并发指南》。 ( 1 ) Java 的线程并发库介绍 Java5 的多线程并有两个大发库在 java.util.concurrent 包及子包中，子包主要的包有一下两个 1) java.util.concurrent 包 (多线程并发库) ➢ java.util.concurrent 包含许多线程安全、测试良好、高性能的并发构建块。不客气地说，创建 java.util.concurrent 的目的就是要实现 Collection 框架对数据结构所执行的并发操作。通过提供一组可靠的、 高性能并发构建块，开发人员可以提高并发类的线程安全、可伸缩性、性能、可读性和可靠性，后面、我们会做介 绍。 ➢ 如果一些类名看起来相似，可能是因为 java.util.concurrent 中的许多概念源自 Doug Lea 的 util.concurrent 库。 2) java.util.concurrent.atomic 包 (多线程的原子性操作提供的工具类) ➢ 查看 atomic 包文档页下面的介绍，它可以对多线程的基本数据、数组中的基本数据和对象中的基本数据 进行多线程的操作（AtomicInteger、AtomicIntegerArray、AtomicIntegerFieldUpDater…） ➢ 通过如下两个方法快速理解 atomic 包的意义：  AtomicInteger 类的 boolean compareAndSet(expectedValue, updateValue);  AtomicIntegerArray 类的 int addAndGet(int i, int delta); ➢ 顺带解释 volatile 类型的作用，需要查看 java 语言规范。  volatile 修饰的变量，线程在每次使用变量的时候，都会读取变量修改后的最的值。（具有可见性）  volatile 没有原子性。 3) java.util.concurrent.lock 包 (多线程的锁机制) 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 120 为锁和等待条件提供一个框架的接口和类，它不同于内置同步和监视器。该框架允许更灵活地使用锁和条件。 本包下有三大接口，下面简单介绍下： ➢ Lock 接口：支持那些语义不同（重入、公平等）的锁规则，可以在非阻塞式结构的上下文（包括 hand- over-hand 和锁重排算法）中使用这些规则。主要的实现是 ReentrantLock。 ➢ ReadWriteLock 接口：以类似方式定义了一些读取者可以共享而写入者独占的锁。此包只提供了一个实 现，即 ReentrantReadWriteLock，因为它适用于大部分的标准用法上下文。但程序员可以创建自己的、 适用于非标准要求的实现。 ➢ Condition 接口：描述了可能会与锁有关联的条件变量。这些变量在用法上与使用 Object.wait 访问的 隐式监视器类似，但提供了更强大的功能。需要特别指出的是，单个 Lock 可能与多个 Condition 对象关 联。为了避免兼容性问题，Condition 方法的名称与对应的 Object 版本中的不同。 ( 2 ) Java 的并发库入门 下面我们将分别介绍 java.util.concurrent 包下的常用类的使用。 1) java.util.concurrent 包 java.util.concurrent 包描述： 在并发编程中很常用的实用工具类。此包包括了几个小的、已标准化的可扩展框架，以及一些提供有用功能 的类。此包下有一些组件，其中包括：  执行程序（线程池）  并发队列  同步器  并发 Collocation 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 121 下面我们将 java.util.concurrent 包下的组件逐一简单介绍： A. 执行程序 ➢ Executors 线程池工厂类 首次我们来说下线程池的作用: 线程池作用就是限制系统中执行线程的数量。 根据系统的环境情况，可以自动或手动设置线程数量，达到运行的最佳效果；少了浪费了系统资源，多了 造成系统拥挤效率不高。用线程池控制线程数量，其他线程 排队等候。一个任务执行完毕，再从队列的中取 最前面的任务开始执行。若队列中没有等待进程，线程池的这一资源处于等待。当一个新任务需要运行时，如 果线程 池中有等待的工作线程，就可以开始运行了；否则进入等待队列。 为什么要用线程池:  减少了创建和销毁线程的次数，每个工作线程都可以被重复利用，可执行多个任务  可以根据系统的承受能力，调整线程池中工作线线程的数目，防止因为因为消耗过多的内存，而把服务 器累趴下(每个线程需要大约 1MB 内存，线程开的越多，消耗的内存也就越大，最后死机) Executors 详解: Java 里面线程池的顶级接口是 Executor，但是严格意义上讲 Executor 并不是一个线程池，而只是一个 执行线程的工具。真正的线程池接口是 ExecutorService。ThreadPoolExecutor 是 Executors 类的底层实 现。我们先介绍下 Executors。 线程池的基本思想还是一种对象池的思想，开辟一块内存空间，里面存放了众多(未死亡)的线程，池中线 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 122 程执行调度由池管理器来处理。当有线程任务时，从池中取一个，执行完成后线程对象归池，这样可以避免反 复创建线程对象所带来的性能开销，节省了系统的资源。 Java5 中并发库中，线程池创建线程大致可以分为下面三种： //创建固定大小的线程池 ExecutorService fPool = Executors.newFixedThreadPool(3); //创建缓存大小的线程池 ExecutorService cPool = Executors.newCachedThreadPool(); //创建单一的线程池 ExecutorService sPool = Executors.newSingleThreadExecutor(); 下面我们通过简单示例来分别说明：  固定大小连接池 import java.util.concurrent.Executors; import java.util.concurrent.ExecutorService; /** * Java 线程：线程池- * * @author Administrator 2009-11-4 23:30:44 */ public class Test { public static void main(String[] args) { //创建一个可重用固定线程数的线程池 ExecutorService pool = Executors.newFixedThreadPool(2); //创建实现了 Runnable 接口对象，Thread 对象当然也实现了 Runnable 接口 Thread t1 = new MyThread(); Thread t2 = new MyThread(); Thread t3 = new MyThread(); Thread t4 = new MyThread(); Thread t5 = new MyThread(); //将线程放入池中进行执行 pool.execute(t1); pool.execute(t2); pool.execute(t3); pool.execute(t4); pool.execute(t5); //关闭线程池 pool.shutdown(); } 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 123 } class MyThread extends Thread{ @Override public void run() { System.out.println(Thread.currentThread().getName()+"正在执行。。。"); } } 运行结果： pool-1-thread-1 正在执行。。。 pool-1-thread-1 正在执行。。。 pool-1-thread-2 正在执行。。。 pool-1-thread-1 正在执行。。。 pool-1-thread-2 正在执行。。。 从上面的运行来看，我们 Thread 类都是在线程池中运行的，线程池在执行 execute 方法来执行 Thread 类 中的 run 方法。不管 execute 执行几次，线程池始终都会使用 2 个线程来处理。不会再去创建出其他线程来处理 run 方法执行。这就是固定大小线程池。  单任务连接池 我们将上面的代码 //创建一个可重用固定线程数的线程池 ExecutorService pool = Executors.newFixedThreadPool(2); 改为： //创建一个使用单个 worker 线程的 Executor，以无界队列方式来运行该线程。 ExecutorService pool = Executors.newSingleThreadExecutor(); 运行结果： pool-1-thread-1 正在执行。。。 pool-1-thread-1 正在执行。。。 pool-1-thread-1 正在执行。。。 pool-1-thread-1 正在执行。。。 pool-1-thread-1 正在执行。。。 运行结果看出，单任务线程池在执行 execute 方法来执行 Thread 类中的 run 方法。不管 execute 执行几次，线程池 始终都会使用单个线程来处理。 补充：在 java 的多线程中，一但线程关闭，就会成为死线程。关闭后死线程就没有办法在启动了。再次启动就会出现 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 124 异常信息：Exception in thread "main" java.lang.IllegalThreadStateException。那么如何解决这个问题呢？ 我们这里就可以使用 Executors.newSingleThreadExecutor()来再次启动一个线程。（面试）  可变连接池 //创建一个可重用固定线程数的线程池 ExecutorService pool = Executors.newFixedThreadPool(2); 改为： //创建一个使用单个 worker 线程的 Executor，以无界队列方式来运行该线程。 ExecutorService pool = Executors.newCachedThreadPool(); 运行结果： pool-1-thread-5 正在执行。。。 pool-1-thread-1 正在执行。。。 pool-1-thread-4 正在执行。。。 pool-1-thread-3 正在执行。。。 pool-1-thread-2 正在执行。。。 运行结果看出，可变任务线程池在执行 execute 方法来执行 Thread 类中的 run 方法。这里 execute 执行多次， 线程池就会创建出多个线程来处理 Thread 类中 run 方法。所有我们看到连接池会根据执行的情况，在程序运行时创 建多个线程来处理，这里就是可变连接池的特点。 那么在上面的三种创建方式，Executors 还可以在执行某个线程时，定时操作。那么下面我们通过代码简单演示 下。  延迟连接池 import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; /** * Java 线程：线程池- * * @author Administrator 2009-11-4 23:30:44 */ public class Test { public static void main(String[] args) { 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 125 //创建一个线程池，它可安排在给定延迟后运行命令或者定期地执行。 ScheduledExecutorService pool = Executors.newScheduledThreadPool(2); //创建实现了 Runnable 接口对象，Thread 对象当然也实现了 Runnable 接口 Thread t1 = new MyThread(); Thread t2 = new MyThread(); Thread t3 = new MyThread(); Thread t4 = new MyThread(); Thread t5 = new MyThread(); //将线程放入池中进行执行 pool.execute(t1); pool.execute(t2); pool.execute(t3); //使用定时执行风格的方法 pool.schedule(t4, 10, TimeUnit.MILLISECONDS); //t4 和 t5 在 10 秒后执行 pool.schedule(t5, 10, TimeUnit.MILLISECONDS); //关闭线程池 pool.shutdown(); } } class MyThread extends Thread { @Override public void run() { System.out.println(Thread.currentThread().getName() + "正在执行。。。"); } } 运行结果： pool-1-thread-1 正在执行。。。 pool-1-thread-2 正在执行。。。 pool-1-thread-1 正在执行。。。 pool-1-thread-1 正在执行。。。 pool-1-thread-2 正在执行。。。 ➢ ExecutorService 执行器服务 java.util.concurrent.ExecutorService 接口表示一个异步执行机制，使我们能够在后台执行任务。因此一 个 ExecutorService 很类似于一个线程池。实际上，存在于 java.util.concurrent 包里的 ExecutorService 实 现就是一个线程池实现。 ExecutorService 例子: 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 126 以下是一个简单的 ExecutorService 例子： //线程工厂类创建出线程池 ExecutorService executorService = Executors.newFixedThreadPool(10); //执行一个线程任务 executorService.execute(new Runnable() { public void run() { System.out.println("Asynchronous task"); } }); //线程池关闭 executorService.shutdown(); 上面代码首先使用 newFixedThreadPool() 工厂方法创建一个 ExecutorService。这里创建了一个十个线程执行任务 的线程池。然后，将一个 Runnable 接口的匿名实现类传递给 execute() 方法。这将导致 ExecutorService 中的某 个线程执行该 Runnable。这里可以看成一个任务分派，示例代码中的任务分派我们可以理解为： 一个线程将一个任务委派给一个 ExecutorService 去异步执行。 一旦该线程将任务委派给 ExecutorService，该线程将继续它自己的执行，独立于该任务的执行。 如下图： Thread1 Operation 1 Operation 2 Operation 3 Operation 4 ExecutorServiceTask Operation 1 Operation 2 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 127 ExecutorService 实现: 既然 ExecutorService 是个接口，如果你想用它的话就得去使用它的实现类之一。 java.util.concurrent 包提供了 ExecutorService 接口的以下实现类：  ThreadPoolExecutor  ScheduledThreadPoolExecutor ExecutorService 创建: ExecutorService 的创建依赖于你使用的具体实现。但是你也可以使用 Executors 工 厂 类 来 创 建 ExecutorService 实例。代码示例： ExecutorService executorService1 = Executors.newSingleThreadExecutor(); //之前 Executors 已介绍 ExecutorService executorService2 = Executors.newFixedThreadPool(10); ExecutorService executorService3 = Executors.newScheduledThreadPool(10); ExecutorService 使用: 有几种不同的方式来将任务委托给 ExecutorService 去执行：  execute(Runnable)  submit(Runnable) 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 128  submit(Callable)  invokeAny(…)  invokeAll(…) 接下来我们挨个看一下这些方法。 ✓ execute(Runnable) execute(Runnable) 方法要求一个 java.lang.Runnable 对象，然后对它进行异步执行。以下是使用 ExecutorService 执行一个 Runnable 的示例： //从 Executors 中获得 ExecutorService ExecutorService executorService = Executors.newSingleThreadExecutor(); //执行 ExecutorService 中的方法 executorService.execute(new Runnable() { public void run() { System.out.println("Asynchronous task"); } }); //线程池关闭 executorService.shutdown(); 特点：没有办法得知被执行的 Runnable 的执行结果。如果有需要的话你得使用一个 Callable(以下将做介 绍)。 ✓ submit(Runnable) submit(Runnable) 方法也要求一个 Runnable 实现类，但它返回一个 Future 对象。这个 Future 对象可 以用来检查 Runnable 是否已经执行完毕。以下是 ExecutorService submit() 示例： //从 Executors 中获得 ExecutorService ExecutorService executorService = Executors.newSingleThreadExecutor(); Future future = executorService.submit(new Runnable() { public void run() { System.out.println("Asynchronous task"); } 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 129 }); future.get(); //获得执行完 run 方法后的返回值，这里使用的 Runnable，所以这里没有返回值，返回的是 null。 executorService.shutdown(); ✓ submit(Runnable) submit(Callable) 方法类似于 submit(Runnable) 方法，除了它所要求的参数类型之外。Callable 实例除了 它的 call() 方法能够返回一个结果之外和一个 Runnable 很相像。Runnable.run() 不能够返回一个结果。 Callable 的结果可以通过 submit(Callable) 方法返回的 Future 对象进行获取。 以下是一个 ExecutorService Callable 示例： //从 Executors 中获得 ExecutorService ExecutorService executorService = Executors.newSingleThreadExecutor(); 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()); executorService.shutdown(); 输出： Asynchronous Callable future.get() = Callable Result ✓ invokeAny() invokeAny() 方法要求一系列的 Callable 或者其子接口的实例对象。调用这个方法并不会返回一个 Future， 但它返回其中一个 Callable 对象的结果。无法保证返回的是哪个 Callable 的结果 – 只能表明其中一个已 执行结束。 如果其中一个任务执行结束(或者抛了一个异常)，其他 Callable 将被取消。以下是示例代码： ExecutorService executorService = Executors.newSingleThreadExecutor(); 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 130 Set> callables = new HashSet>(); callables.add(new Callable() { public String call() throws Exception { return "Task 1"; } }); callables.add(new Callable() { public String call() throws Exception { return "Task 2"; } }); callables.add(new Callable() { public String call() throws Exception { return "Task 3"; } }); String result = executorService.invokeAny(callables); System.out.println("result = " + result); executorService.shutdown(); 上述代码将会打印出给定 Callable 集合中的一个的执行结果。我自己试着执行了它几次，结果始终在变。有 时是 “Task 1″，有时是 “Task 2″ 等等。 ✓ invokeAll() invokeAll() 方法将调用你在集合中传给 ExecutorService 的所有 Callable 对象。invokeAll() 返回一系列 的 Future 对象，通过它们你可以获取每个 Callable 的执行结果。 记住，一个任务可能会由于一个异常而结束，因此它可能没有 "成功"。无法通过一个 Future 对象来告知我 们是两种结束中的哪一种。 以下是一个代码示例： ExecutorService executorService = Executors.newSingleThreadExecutor(); Set> callables = new HashSet>(); 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 131 callables.add(new Callable() { public String call() throws Exception { return "Task 1"; } }); callables.add(new Callable() { public String call() throws Exception { return "Task 2"; } }); callables.add(new Callable() { public String call() throws Exception { return "Task 3"; } }); List> futures = executorService.invokeAll(callables); for(Future future : futures){ System.out.println("future.get = " + future.get()); } executorService.shutdown(); 输出结果： future.get = Task 3 future.get = Task 1 future.get = Task 2 Executors 关闭: 使用 shutdown 和 shutdownNow 可以关闭线程池 两者的区别： shutdown 只是将空闲的线程 interrupt() 了，shutdown（）之前提交的任务可以继续执行直到结束。 shutdownNow 是 interrupt 所有线程， 因此大部分线程将立刻被中断。之所以是大部分，而不是全部 ， 是因为 interrupt()方法能力有限。 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 132 ➢ ThreadPoolExecutor 线程池执行者 java.util.concurrent.ThreadPoolExecutor 是 ExecutorService 接口的一个实现。ThreadPoolExecutor 使用 其内部池中的线程执行给定任务(Callable 或者 Runnable)。 ThreadPoolExecutor 包含的线程池能够包含不同数量的线程。池中线程的数量由以下变量决定：  corePoolSize  maximumPoolSize 当一个任务委托给线程池时，如果池中线程数量低于 corePoolSize，一个新的线程将被创建，即使池中可能尚有 空闲线程。如果内部任务队列已满，而且有至少 corePoolSize 正在运行，但是运行线程的数量低于 maximumPoolSize，一个新的线程将被创建去执行该任务。 ThreadPoolExecutor 图解： Thread Pool Executor Task Queue Task Task Task Thread maxPoolSize corePoolSize Thread Thread Thread 创建 ThreadPoolExecutor: int corePoolSize = 5; 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 133 int maxPoolSize = 10; long keepAliveTime = 5000; ExecutorService threadPoolExecutor = new ThreadPoolExecutor( corePoolSize, maxPoolSize, keepAliveTime, TimeUnit.MILLISECONDS, new LinkedBlockingQueue() ); 构造方法参数列表解释： corePoolSize - 池中所保存的线程数，包括空闲线程。 maximumPoolSize - 池中允许的最大线程数。 keepAliveTime - 当线程数大于核心时，此为终止前多余的空闲线程等待新任务的最长时间。 unit - keepAliveTime 参数的时间单位。 workQueue - 执行前用于保持任务的队列。此队列仅保持由 execute 方法提交的 Runnable 任务。 ➢ ScheduledPoolExecutor 定时线程池执行者 java.util.concurrent.ScheduledExecutorService 是一个 ExecutorService， 它能够将任务延后执行，或者间隔 固定时间多次执行。 任务由一个工作者线程异步执行，而不是由提交任务给 ScheduledExecutorService 的那个线 程执行。 ScheduledPoolExecutor 例子: 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 秒后执行 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 134 首先一个内置 5 个线程的 ScheduledExecutorService 被创建。之后一个 Callable 接口的匿名类示例被创建 然后传递给 schedule() 方法。后边的俩参数定义了 Callable 将在 5 秒钟之后被执行。 ScheduledExecutorService 的实现: ScheduledExecutorService 是一个接口，你要用它的话就得使用 java.util.concurrent 包里对它的某个实现类。 ScheduledExecutorService 具有以下实现类：ScheduledThreadPoolExecutor 创建一个 ScheduledExecutorService: 如何创建一个 ScheduledExecutorService 取决于你采用的它的实现类。但是你也可以使用 Executors 工厂类 来创建一个 ScheduledExecutorService 实例。比如： ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(5); ScheduledExecutorService 的使用: 一旦你创建了一个 ScheduledExecutorService，你可以通过调用它的以下方法：  schedule (Callable task, long delay, TimeUnit timeunit)  schedule (Runnable task, long delay, TimeUnit timeunit)  scheduleAtFixedRate (Runnable, long initialDelay, long period, TimeUnit timeunit)  scheduleWithFixedDelay (Runnable, long initialDelay, long period, TimeUnit timeunit) 下面我们就简单看一下这些方法。 ✓ schedule (Callable task, long delay, TimeUnit timeunit) ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(5); ScheduledFuture scheduledFuture = scheduledExecutorService.schedule(new Callable() { 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 135 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! ✓ schedule (Runnable task, long delay, TimeUnit timeunit) 这一方法规划一个任务将被定期执行。该任务将会在首个 initialDelay 之后得到执行，然后每个 period 时间之 后重复执行。 如果给定任务的执行抛出了异常，该任务将不再执行。如果没有任何异常的话，这个任务将会持续循环执行到 ScheduledExecutorService 被关闭。 如果一个任务占用了比计划的时间间隔更长的时候，下一次执行将在当前执行结束执行才开始。计划任务在同一 时间不会有多个线程同时执行。 ✓ scheduleAtFixedRate (Runnable, long initialDelay, long period, TimeUnit timeunit) 这一方法规划一个任务将被定期执行。该任务将会在首个 initialDelay 之后得到执行，然后每个 period 时间之 后重复执行。 如果给定任务的执行抛出了异常，该任务将不再执行。如果没有任何异常的话，这个任务将会持续循环执行到 ScheduledExecutorService 被关闭。 如果一个任务占用了比计划的时间间隔更长的时候，下一次执行将在当前执行结束执行才开始。计划任务在同一 时间不会有多个线程同时执行。 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 136 ✓ scheduleWithFixedDelay (Runnable, long initialDelay, long period, TimeUnit timeunit) 除了 period 有不同的解释之外这个方法和 scheduleAtFixedRate() 非常像。scheduleAtFixedRate() 方法中， period 被解释为前一个执行的开始和下一个执行的开始之间的间隔时间。而在本方法中，period 则被解释为前一个 执行的结束和下一个执行的结束之间的间隔。因此这个延迟是执行结束之间的间隔，而不是执行开始之间的间隔。 ScheduledExecutorService 的关闭: 正如 ExecutorService，在你使用结束之后你需要把 ScheduledExecutorService 关闭掉。否则他将导致 JVM 继续运行，即使所有其他线程已经全被关闭。 你 可 以 使 用 从 ExecutorService 接 口 继 承 来 的 shutdown() 或 shutdownNow() 方 法 将 ScheduledExecutorService 关闭。参见 ExecutorService 关闭部分以获取更多信息。 ➢ ForkJoinPool 合并和分叉（线程池） ForkJoinPool 在 Java 7 中被引入。它和 ExecutorService 很相似，除了一点不同。ForkJoinPool 让我们可以 很方便地把任务分裂成几个更小的任务，这些分裂出来的任务也将会提交给 ForkJoinPool。任务可以继续分割成更小 的子任务，只要它还能分割。可能听起来有些抽象，因此本节中我们将会解释 ForkJoinPool 是如何工作的，还有任 务分割是如何进行的。 合并和分叉的解释: 在我们开始看 ForkJoinPool 之前我们先来简要解释一下分叉和合并的原理。分叉和合并原理包含两个递归进行 的步骤。两个步骤分别是分叉步骤和合并步骤。 分叉: 一个使用了分叉和合并原理的任务可以将自己分叉(分割)为更小的子任务，这些子任务可以被并发执行。如下图所 示： Task Task Task Task Task Task Task Fork 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 137 通过把自己分割成多个子任务，每个子任务可以由不同的 CPU 并行执行，或者被同一个 CPU 上的不同线程执 行。只有当给的任务过大，把它分割成几个子任务才有意义。把任务分割成子任务有一定开销，因此对于小型任务，这 个分割的消耗可能比每个子任务并发执行的消耗还要大。 什么时候把一个任务分割成子任务是有意义的，这个界限也称作一个阀值。这要看每个任务对有意义阀值的决定。 很大程度上取决于它要做的工作的种类。 合并: 当一个任务将自己分割成若干子任务之后，该任务将进入等待所有子任务的结束之中。一旦子任务执行结束，该 任务可以把所有结果合并到同一个结果。图示如下： 当然，并非所有类型的任务都会返回一个结果。如果这个任务并不返回一个结果，它只需等待所有子任务执行完 毕。也就不需要结果的合并啦。 Task Task Task Task Task Task Task Join 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 138 所以我们可以将 ForkJoinPool 是一个特殊的线程池，它的设计是为了更好的配合 分叉-和-合并 任务分割的工 作。ForkJoinPool 也在 java.util.concurrent 包中，其完整类名为 java.util.concurrent.ForkJoinPool。 创建一个 ForkJoinPool: 你可以通过其构造子创建一个 ForkJoinPool。作为传递给 ForkJoinPool 构造子的一个参数，你可以定义你期望 的并行级别。并行级别表示你想要传递给 ForkJoinPool 的任务所需的线程或 CPU 数量。以下是一个 ForkJoinPool 示例： //创建了一个并行级别为 4 的 ForkJoinPool ForkJoinPool forkJoinPool = new ForkJoinPool(4); 提交任务到 ForkJoinPool: 就像提交任务到 ExecutorService 那样，把任务提交到 ForkJoinPool。你可以提交两种类型的任务。一种是没 有任何返回值的(一个 “行动”)，另一种是有返回值的(一个”任务”)。这两种类型分别由 RecursiveAction 和 RecursiveTask 表示。接下来介绍如何使用这两种类型的任务，以及如何对它们进行提交。 RecursiveAction: RecursiveAction 是一种没有任何返回值的任务。它只是做一些工作，比如写数据到磁盘，然后就退出了。一个 RecursiveAction 可以把自己的工作分割成更小的几块，这样它们可以由独立的线程或者 CPU 执行。你可以通过继 承来实现一个 RecursiveAction。示例如下： import java.util.ArrayList; import java.util.List; import java.util.concurrent.RecursiveAction; public class MyRecursiveAction extends RecursiveAction { private long workLoad = 0; public MyRecursiveAction(long workLoad) { this.workLoad = workLoad; } 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 139 @Override protected void compute() { //if work is above threshold, break tasks up into smaller tasks //翻译：如果工作超过门槛，把任务分解成更小的任务 if(this.workLoad > 16) { System.out.println("Splitting workLoad : " + this.workLoad); List subtasks = new ArrayList(); subtasks.addAll(createSubtasks()); for(RecursiveAction subtask : subtasks){ subtask.fork(); } } else { System.out.println("Doing workLoad myself: " + this.workLoad); } } private List createSubtasks() { List subtasks = new ArrayList(); MyRecursiveAction subtask1 = new MyRecursiveAction(this.workLoad / 2); MyRecursiveAction subtask2 = new MyRecursiveAction(this.workLoad / 2); subtasks.add(subtask1); subtasks.add(subtask2); return subtasks; } } 例子很简单。MyRecursiveAction 将一个虚构的 workLoad 作为参数传给自己的构造子。如果 workLoad 高于 一个特定阀值，该工作将被分割为几个子工作，子工作继续分割。如果 workLoad 低于特定阀值，该工作将由 MyRecursiveAction 自己执行。你可以这样规划一个 MyRecursiveAction 的执行： 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 140 //创建了一个并行级别为 4 的 ForkJoinPool ForkJoinPool forkJoinPool = new ForkJoinPool(4); //创建一个没有返回值的任务 MyRecursiveAction myRecursiveAction = new MyRecursiveAction(24); //ForkJoinPool 执行任务 forkJoinPool.invoke(myRecursiveAction); 运行结果： Splitting workLoad : 24 Doing workLoad myself: 12 RecursiveTask: RecursiveTask 是一种会返回结果的任务。它可以将自己的工作分割为若干更小任务，并将这些子任务的执行结 果合并到一个集体结果。可以有几个水平的分割和合并。以下是一个 RecursiveTask 示例： import java.util.ArrayList; import java.util.List; import java.util.concurrent.RecursiveTask; public class MyRecursiveTask extends RecursiveTask { private long workLoad = 0; public MyRecursiveTask(long workLoad) { this.workLoad = workLoad; } protected Long compute() { //if work is above threshold, break tasks up into smaller tasks if(this.workLoad > 16) { System.out.println("Splitting workLoad : " + this.workLoad); List subtasks = new ArrayList(); subtasks.addAll(createSubtasks()); for(MyRecursiveTask subtask : subtasks){ subtask.fork(); 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 141 } long result = 0; for(MyRecursiveTask subtask : subtasks) { result += subtask.join(); } return result; } else { System.out.println("Doing workLoad myself: " + this.workLoad); return workLoad * 3; } } private List createSubtasks() { List subtasks = new ArrayList(); MyRecursiveTask subtask1 = new MyRecursiveTask(this.workLoad / 2); MyRecursiveTask subtask2 = new MyRecursiveTask(this.workLoad / 2); subtasks.add(subtask1); subtasks.add(subtask2); return subtasks; } } 除了有一个结果返回之外，这个示例和 RecursiveAction 的 例 子 很 像 。 MyRecursiveTask 类 继 承 自 RecursiveTask，这也就意味着它将返回一个 Long 类型的结果。 MyRecursiveTask 示例也会将工作分割为子任务，并通过 fork() 方法对这些子任务计划执行。此外，本示例还 通过调用每个子任务的 join() 方法收集它们返回的结果。子任务的结果随后被合并到一个更大的结果，并最终将其返 回。对于不同级别的递归，这种子任务的结果合并可能会发生递归。 你可以这样规划一个 RecursiveTask： //创建了一个并行级别为 4 的 ForkJoinPool ForkJoinPool forkJoinPool = new ForkJoinPool(4); //创建一个有返回值的任务 MyRecursiveTask myRecursiveTask = new MyRecursiveTask(128); 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 142 //线程池执行并返回结果 long mergedResult = forkJoinPool.invoke(myRecursiveTask); System.out.println("mergedResult = " + mergedResult); 注意: ForkJoinPool.invoke() 方法的调用来获取最终执行结果的。 B. 并发队列-阻塞队列 常用的并发队列有阻塞队列和非阻塞队列，前者使用锁实现，后者则使用 CAS 非阻塞算法实现。 PS：至于非阻塞队列是靠 CAS 非阻塞算法，在这里不再介绍，大家只用知道，Java 非阻塞队列是使用 CAS 算法来实现的就可 以。感兴趣的童鞋可以维基网上自行学习. 下面我们先介绍阻塞队列。 阻塞队列: 阻塞队列 (BlockingQueue)是 Java util.concurrent 包下重要的数据结构，BlockingQueue 提供了线程安全的队 列访问方式：当阻塞队列进行插入数据时，如果队列已满，线程将会阻塞等待直到队列非满；从阻塞队列取数据时，如 果队列已空，线程将会阻塞等待直到队列非空。并发包下很多高级同步类的实现都是基于 BlockingQueue 实现的。 ➢ BlockingQueue 阻塞队列 BlockingQueue 通常用于一个线程生产对象，而另外一个线程消费这些对象的场景。下图是对这个原理的阐述： Thread 1 Thread 2 BlockingQueue Put Take 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 143 一个线程往里边放，另外一个线程从里边取的一个 BlockingQueue。 一个线程将会持续生产新对象并将其插入到队列之中，直到队列达到它所能容纳的临界点。也就是说，它是有限 的。如果该阻塞队列到达了其临界点，负责生产的线程将会在往里边插入新对象时发生阻塞。它会一直处于阻塞之中， 直到负责消费的线程从队列中拿走一个对象。负责消费的线程将会一直从该阻塞队列中拿出对象。如果消费线程尝试 去从一个空的队列中提取对象的话，这个消费线程将会处于阻塞之中，直到一个生产线程把一个对象丢进队列。 BlockingQueue 的方法: BlockingQueue 具有 4 组不同的方法用于插入、移除以及对队列中的元素进行检查。如果请求的操作不能得到 立即执行的话，每个方法的表现也不同。这些方法如下： 阻塞队列提供了四种处理方法: 四组不同的行为方式解释： 抛异常：如果试图的操作无法立即执行，抛一个异常。 特定值：如果试图的操作无法立即执行，返回一个特定的值(常常是 true / false)。 阻塞：如果试图的操作无法立即执行，该方法调用将会发生阻塞，直到能够执行。 超时：如果试图的操作无法立即执行，该方法调用将会发生阻塞，直到能够执行，但等待时间不会超过给定 值。返回一个特定值以告知该操作是否成功(典型的是 true / false)。 方法\处理方式 抛出异常 返回特殊值 一直阻塞 超时退出 插入方法 add(e) offer(e) put(e) offer(e,time,unit) 移除方法 remove() poll() take() poll(time,unit) 检查方法 element() peek() 不可用 不可用 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 144 无法向一个 BlockingQueue 中插入 null。如果你试图插入 null，BlockingQueue 将会抛出一个 NullPointerException. BlockingQueue 的实现类: BlockingQueue 是个接口，你需要使用它的实现之一来使用 BlockingQueue，Java.util.concurrent 包下具有以 下 BlockingQueue 接口的实现类：  ArrayBlockingQueue：ArrayBlockingQueue 是一个有界的阻塞队列，其内部实现是将对象放到一个 数组里。有界也就意味着，它不能够存储无限多数量的元素。它有一个同一时间能够存储元素数量的上限。你 可以在对其初始化的时候设定这个上限，但之后就无法对这个上限进行修改了(译者注：因为它是基于数组实 现的，也就具有数组的特性：一旦初始化，大小就无法修改)。  DelayQueue：DelayQueue 对元素进行持有直到一个特定的延迟到期。注入其中的元素必须实现 java.util.concurrent.Delayed 接口。  LinkedBlockingQueue：LinkedBlockingQueue 内部以一个链式结构(链接节点)对其元素进行存储。如 果需要的话，这一链式结构可以选择一个上限。如果没有定义上限，将使用 Integer.MAX_VALUE 作为上限。  PriorityBlockingQueue ： PriorityBlockingQueue 是一个无界的并发队列。它使用了和类 java.util.PriorityQueue 一样的排序规则。你无法向这个队列中插入 null 值 。 所 有 插 入 到 PriorityBlockingQueue 的元素必须实现 java.lang.Comparable 接口。因此该队列中元素的排序就取决于 你自己的 Comparable 实现。  SynchronousQueue：SynchronousQueue 是一个特殊的队列，它的内部同时只能够容纳单个元素。 如果该队列已有一元素的话，试图向队列中插入一个新元素的线程将会阻塞，直到另一个线程将该元素从队 列中抽走。同样，如果该队列为空，试图向队列中抽取一个元素的线程将会阻塞，直到另一个线程向队列中插 入了一条新的元素。据此，把这个类称作一个队列显然是夸大其词了。它更多像是一个汇合点。 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 145 ➢ ArrayBlockingQueue 阻塞队列 ArrayBlockingQueue 类图 如上图 ArrayBlockingQueue 内部有个数组 items 用来存放队列元素，putindex 下标标示入队元素下标， takeIndex 是出队下标，count 统计队列元素个数，从定义可知道并没有使用 volatile 修饰，这是因为访问这些变量使 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 146 用都是在锁块内，并不存在可见性问题。另外有个独占锁 lock 用来对出入队操作加锁，这导致同时只有一个线程可以 访问入队出队，另外 notEmpty，notFull 条件变量用来进行出入队的同步。 另外构造函数必须传入队列大小参数，所以为有界队列，默认是 Lock 为非公平锁。 public ArrayBlockingQueue(int capacity) { this(capacity, false); } public ArrayBlockingQueue(int capacity, boolean fair) { if (capacity <= 0) throw new IllegalArgumentException(); this.items = new Object[capacity]; lock = new ReentrantLock(fair); notEmpty = lock.newCondition(); notFull = lock.newCondition(); } ps： 所谓公平锁：就是在并发环境中，每个线程在获取锁时会先查看此锁维护的等待队列，如果为空，或者当前线 程线程是等待队列的第一个，就占有锁，否则就会加入到等待队列中，以后会按照 FIFO 的规则从队列中取到 自己。 非公平锁：比较粗鲁，上来就直接尝试占有锁，如果尝试失败，就再采用类似公平锁那种方式 ArrayBlockingQueue 方法 ✓ offer 方法 在队尾插入元素，如果队列满则返回 false，否者入队返回 true。 public boolean offer(E e) { //e 为 null，则抛出 NullPointerException 异常 checkNotNull(e); //获取独占锁 final ReentrantLock lock = this.lock; lock.lock(); try { 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 147 //如果队列满则返回 false if (count == items.length) return false; else { //否者插入元素 insert(e); return true; } } finally { //释放锁 lock.unlock(); } } private void insert(E x) { //元素入队 items[putIndex] = x; //计算下一个元素应该存放的下标 putIndex = inc(putIndex); ++count; notEmpty.signal(); } //循环队列，计算下标 final int inc(int i) { return (++i == items.length) ? 0 : i; } 这里由于在操作共享变量前加了锁，所以不存在内存不可见问题，加过锁后获取的共享变量都是从主内存获取的， 而不是在 CPU 缓存或者寄存器里面的值，释放锁后修改的共享变量值会刷新会主内存中。 另外这个队列是使用循环数组实现，所以计算下一个元素存放下标时候有些特殊。另外 insert 后调用 notEmpty.signal();是为了激活调用 notEmpty.await()阻塞后放入 notEmpty 条件队列中的线程。 ✓ Put 操作 在队列尾部添加元素，如果队列满则等待队列有空位置插入后返回。 public void put(E e) throws InterruptedException { 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 148 checkNotNull(e); final ReentrantLock lock = this.lock; //获取可被中断锁 lock.lockInterruptibly(); try { //如果队列满，则把当前线程放入 notFull 管理的条件队列 while (count == items.length) notFull.await(); //插入元素 insert(e); } finally { lock.unlock(); } } 需要注意的是如果队列满了那么当前线程会阻塞，知道出队操作调用了 notFull.signal 方法激活该线程。代码逻 辑很简单，但是这里需要思考一个问题为啥调用 lockInterruptibly 方法而不是 Lock 方法。我的理解是因为调用了条 件变量的 await()方法，而 await()方法会在中断标志设置后抛出 InterruptedException 异常后退出，所以还不如在加 锁时候先看中断标志是不是被设置了，如果设置了直接抛出 InterruptedException 异常，就不用再去获取锁了。然后 看了其他并发类里面凡是调用了 await 的方法获取锁时候都是使用的 lockInterruptibly 方法而不是 Lock 也验证了这 个想法。 ✓ Poll 操作 从队头获取并移除元素，队列为空，则返回 null。 public E poll() { final ReentrantLock lock = this.lock; lock.lock(); try { //当前队列为空则返回 null,否者 return (count == 0) ? null : extract(); } finally { lock.unlock(); } } 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 149 private E extract() { final Object[] items = this.items; //获取元素值 E x = this.cast(items[takeIndex]); //数组中值值为 null; items[takeIndex] = null; //队头指针计算，队列元素个数减一 takeIndex = inc(takeIndex); --count; //发送信号激活 notFull 条件队列里面的线程 notFull.signal(); return x; } ✓ Take 操作 从队头获取元素，如果队列为空则阻塞直到队列有元素。 public E take() throws InterruptedException { final ReentrantLock lock = this.lock; lock.lockInterruptibly(); try { //队列为空，则等待，直到队列有元素 while (count == 0) notEmpty.await(); return extract(); } finally { lock.unlock(); } } 需要注意的是如果队列为空，当前线程会被挂起放到 notEmpty 的条件队列里面，直到入队操作执行调用 notEmpty.signal 后当前线程才会被激活，await 才会返回。 ✓ Peek 操作 返回队列头元素但不移除该元素，队列为空，返回 null。 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 150 public E peek() { final ReentrantLock lock = this.lock; lock.lock(); try { //队列为空返回 null,否者返回头元素 return (count == 0) ? null : itemAt(takeIndex); } finally { lock.unlock(); } } final E itemAt(int i) { return this.cast(items[i]); } ✓ Size 操作 获取队列元素个数，非常精确因为计算 size 时候加了独占锁，其他线程不能入队或者出队或者删除元素。 public int size() { final ReentrantLock lock = this.lock; lock.lock(); try { return count; } finally { lock.unlock(); } } ✓ ArrayBlockingQueue 小结 ArrayBlockingQueue 通过使用全局独占锁实现同时只能有一个线程进行入队或者出队操作，这个锁的粒度比较 大，有点类似在方法上添加 synchronized 的意味。其中 offer,poll 操作通过简单的加锁进行入队出队操作，而 put,take 则使用了条件变量实现如果队列满则等待，如果队列空则等待，然后分别在出队和入队操作中发送信号激活等待线程 实现同步。另外相比 LinkedBlockingQueue，ArrayBlockingQueue 的 size 操作的结果是精确的，因为计算前加了 全局锁。 ArrayBlockingQueue 示例 需求：在多线程操作下，一个数组中最多只能存入 3 个元素。多放入不可以存入数组，或等待某线程对数组中某 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 151 个元素取走才能放入，要求使用 java 的多线程来实现。（面试） 代码实现： public class BlockingQueueTest { public static void main(String[] args) { final BlockingQueue queue = new ArrayBlockingQueue(3); for(int i=0;i<2;i++){ new Thread(){ public void run(){ while(true){ try { Thread.sleep((long)(Math.random()*1000)); System.out.println(Thread.currentThread().getName() + "准备放数据!"); queue.put(1); System.out.println(Thread.currentThread().getName() + "已经放了数据，" + "队列目前有" + queue.size() + "个数据"); } catch (InterruptedException e) { e.printStackTrace(); } } } }.start(); } new Thread(){ public void run(){ while(true){ try { //将此处的睡眠时间分别改为 100 和 1000，观察运行结果 Thread.sleep(100); System.out.println(Thread.currentThread().getName() + "准备取数据!"); System.err.println(queue.take()); System.out.println(Thread.currentThread().getName() + "已经取走数据，" + "队列目前有" + queue.size() + "个数据"); } catch (InterruptedException e) { e.printStackTrace(); } } 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 152 } }.start(); } } 输出结果： Thread-0 准备放数据! Thread-0 已经放了数据，队列目前有 1 个数据 Thread-0 准备放数据! Thread-0 已经放了数据，队列目前有 2 个数据 Thread-1 准备放数据! Thread-1 已经放了数据，队列目前有 3 个数据 Thread-2 准备取数据! Thread-2 已经取走数据，队列目前有 3 个数据 Thread-0 准备放数据! Thread-1 准备放数据! Thread-2 准备取数据! Thread-2 已经取走数据，队列目前有 3 个数据 ………………… ➢ LinkedBlockingQueue 阻塞队列 LinkedBlockingQueue 类图 LinkedBlockingQueue 中也有两个 Node 分别用来存放首尾节点，并且里面有个初始值为 0 的原子变量 count 用来记录队列元素个数，另外里面有两个 ReentrantLock 的独占锁，分别用来控制元素入队和出队加锁，其中 takeLock 用来控制同时只有一个线程可以从队列获取元素，其他线程必须等待，putLock 控制同时只能有一个线程可以获取锁 去添加元素，其他线程必须等待。另外 notEmpty 和 notFull 用来实现入队和出队的同步。 另外由于出入队是两个非 公平独占锁，所以可以同时又一个线程入队和一个线程出队，其实这个是个生产者-消费者模型，如下类图： 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 153 /** 通过 take 取出进行加锁、取出 */ private final ReentrantLock takeLock = new ReentrantLock(); /** 等待中的队列等待取出 */ private final Condition notEmpty = takeLock.newCondition(); /*通过 put 放置进行加锁、放置*/ private final ReentrantLock putLock = new ReentrantLock(); /** 等待中的队列等待放置 */ private final Condition notFull = putLock.newCondition(); /* 记录集合中的个数（计数器） */ private final AtomicInteger count = new AtomicInteger(0); 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 154 队列的容量： //队列初始容量，Integer 最大值 public static final int MAX_VALUE = 0x7fffffff; public LinkedBlockingQueue() { this(Integer.MAX_VALUE); } public LinkedBlockingQueue(int capacity) { if (capacity <= 0) throw new IllegalArgumentException(); this.capacity = capacity; //初始化首尾节点 last = head = new Node(null); } 如图默认队列容量为 0x7fffffff;用户也可以自己指定容量。 LinkedBlockingQueue 方法 ps：下面介绍 LinkedBlockingQueue 用到很多 Lock 对象。详细可以查找 Lock 对象的介绍 ✓ 带时间的 Offer 操作-生产者 在 ArrayBlockingQueue 中已经简单介绍了 Offer()方法，LinkedBlocking 的 Offer 方法类似，在此就不过多去 介绍。这次我们从介绍下带时间的 Offer 方法 public boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException { //空元素抛空指针异常 if (e == null) throw new NullPointerException(); long nanos = unit.toNanos(timeout); int c = -1; final ReentrantLock putLock = this.putLock; final AtomicInteger count = this.count; //获取可被中断锁，只有一个线程克获取 putLock.lockInterruptibly(); try { //如果队列满则进入循环 while (count.get() == capacity) { 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 155 //nanos<=0 直接返回 if (nanos <= 0) return false; //否者调用 await 进行等待，超时则返回<=0（1） nanos = notFull.awaitNanos(nanos); } //await 在超时时间内返回则添加元素（2） enqueue(new Node(e)); c = count.getAndIncrement(); //队列不满则激活其他等待入队线程（3） if (c + 1 < capacity) notFull.signal(); } finally { //释放锁 putLock.unlock(); } //c==0 说明队列里面有一个元素，这时候唤醒出队线程（4） if (c == 0) signalNotEmpty(); return true; } private void enqueue(Node node) { last = last.next = node; } private void signalNotEmpty() { final ReentrantLock takeLock = this.takeLock; takeLock.lock(); try { notEmpty.signal(); } finally { takeLock.unlock(); } } ✓ 带时间的 poll 操作-消费者 获取并移除队首元素，在指定的时间内去轮询队列看有没有首元素有则返回，否者超时后返回 null。 public E poll(long timeout, TimeUnit unit) throws InterruptedException { E x = null; 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 156 int c = -1; long nanos = unit.toNanos(timeout); final AtomicInteger count = this.count; final ReentrantLock takeLock = this.takeLock; //出队线程获取独占锁 takeLock.lockInterruptibly(); try { //循环直到队列不为空 while (count.get() == 0) { //超时直接返回 null if (nanos <= 0) return null; nanos = notEmpty.awaitNanos(nanos); } //出队，计数器减一 x = dequeue(); c = count.getAndDecrement(); //如果出队前队列不为空则发送信号，激活其他阻塞的出队线程 if (c > 1) notEmpty.signal(); } finally { //释放锁 takeLock.unlock(); } //当前队列容量为最大值-1 则激活入队线程。 if (c == capacity) signalNotFull(); return x; } 首先获取独占锁，然后进入循环当当前队列有元素才会退出循环，或者超时了，直接返回 null。 超时前退出循环后，就从队列移除元素，然后计数器减去一，如果减去 1 前队列元素大于 1 则说明当前移除后队 列还有元素，那么就发信号激活其他可能阻塞到当前条件信号的线程。 最后如果减去 1 前队列元素个数=最大值，那么移除一个后会腾出一个空间来，这时候可以激活可能存在的入队 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 157 阻塞线程。 ✓ put 操作-生产者 与带超时时间的 poll 类似不同在于 put 时候如果当前队列满了它会一直等待其他线程调用 notFull.signal 才会被 唤醒。 ✓ take 操作-消费者 与带超时时间的 poll 类似不同在于 take 时候如果当前队列空了它会一直等待其他线程调用 notEmpty.signal()才 会被唤醒。 ✓ size 操作-消费者 当前队列元素个数，如代码直接使用原子变量 count 获取。 public int size() { return count.get(); } ✓ peek 操作 获取但是不移除当前队列的头元素，没有则返回 null。 public E peek() { //队列空，则返回 null if (count.get() == 0) return null; final ReentrantLock takeLock = this.takeLock; takeLock.lock(); try { Node first = head.next; if (first == null) return null; else return first.item; } finally { takeLock.unlock(); } } ✓ remove 操作 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 158 删除队列里面的一个元素，有则删除返回 true，没有则返回 false，在删除操作时候由于要遍历队列所以加了双重 锁，也就是在删除过程中不允许入队也不允许出队操作。 public boolean remove(Object o) { if (o == null) return false; //双重加锁 fullyLock(); try { //遍历队列找则删除返回 true for (Node trail = head, p = trail.next; p != null; trail = p, p = p.next) { if (o.equals(p.item)) { unlink(p, trail); return true; } } //找不到返回 false return false; } finally { //解锁 fullyUnlock(); } } void fullyLock() { putLock.lock(); takeLock.lock(); } void fullyUnlock() { takeLock.unlock(); putLock.unlock(); } void unlink(Node p, Node trail) { p.item = null; trail.next = p.next; if (last == p) 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 159 last = trail; //如果当前队列满，删除后，也不忘记最快的唤醒等待的线程 if (count.getAndDecrement() == capacity) notFull.signal(); } ✓ 开源框架的使用 tomcat 中任务队列 TaskQueue。 类结构图： 可知 TaskQueue 继承了 LinkedBlockingQueue 并且泛化类型固定了为 Runnalbe.重写了 offer,poll，take 方法。 tomcat 中有个线程池 ThreadPoolExecutor，在 NIOEndPoint 中当 acceptor 线程接受到请求后，会把任务放入队 列，然后 poller 线程从队列里面获取任务，然后就把任务放入线程池执行。这个 ThreadPoolExecutor 中的的一个参 数就是 TaskQueue。 先看看 ThreadPoolExecutor 的参数如果是普通 LinkedBlockingQueue 是怎么样的执行逻辑： 当调用线程池方法 execute() 方法添加一个任务时： 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 160  如果当前运行的线程数量小于 corePoolSize，则创建新线程运行该任务  如果当前运行的线程数量大于或等于 corePoolSize，则将这个任务放入阻塞队列。  如果当前队列满了，并且当前运行的线程数量小于 maximumPoolSize，则创建新线程运行该任务；  如果当前队列满了，并且当前运行的线程数量大于或等于 maximumPoolSize，那么线程池将会抛出 RejectedExecutionException 异常。 如果线程执行完了当前任务，那么会去队列里面获取一个任务来执行，如果任务执行完了，并且当前线程数大于 corePoolSize，那么会根据线程空闲时间 keepAliveTime 回收一些线程保持线程池 corePoolSize 个线程。 首先看下线程池中 exectue 添加任务时候的逻辑： public void execute(Runnable command) { if (command == null) throw new NullPointerException(); //当前工作线程个数小于 core 个数则开新线程执行（1） int c = ctl.get(); if (workerCountOf(c) < corePoolSize) { if (addWorker(command, true)) return; c = ctl.get(); } //放入队列（2） if (isRunning(c) && workQueue.offer(command)) { int recheck = ctl.get(); if (! isRunning(recheck) && remove(command)) reject(command); else if (workerCountOf(recheck) == 0) addWorker(null, false); } //如果队列满了则开新线程，但是个数要不超过最大值，超过则返回 false //然后执行 reject handler（3） else if (!addWorker(command, false)) reject(command); } 可知当当前工作线程个数为 corePoolSize 后，如果在来任务会把任务添加到队列，队列满了或者入队失败了则开 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 161 启新线程。 然后看看 TaskQueue 中重写的 offer 方法的逻辑： public boolean offer(Runnable o) { // 如果 parent 为 null 则直接调用父类方法 if (parent==null) return super.offer(o); //如果当前线程池中线程个数达到最大，则无条件调用父类方法 if (parent.getPoolSize() == parent.getMaximumPoolSize()) return super.offer(o); //如果当前提交的任务小于当前线程池线程数，说明线程用不完，没必要重新开线程 if (parent.getSubmittedCount()<(parent.getPoolSize())) return super.offer(o); //如果当前线程池线程个数>core 个数但是小于最大个数，则开新线程代替放入队列 if (parent.getPoolSize()0 因为可能多个线程 put 了，所以调用 dequeue 从队列获取元素（这时候一定可以 获取到），然后调用 c = count.getAndDecrement() 把当前计数返回后并减去 1，如果 c>1 说明当前队列还有其他 元素，那么就调用 notEmpty.signal()去激活 notEmpty 的条件队列里面的其他阻塞线程。 考虑当队列满的时候： 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 163 当队列满的时候调用 put 方法时候，会由于 notFull.await()当前线程被阻塞放入 notFull 管理的条件队列里面， 同理可能会有多个调用 put 方法的线程都放到了 notFull 的条件队列里面。 这时候如果有一个线程调用了 take 方法,调用 dequeue()出队一个元素，c = count.getAndDecrement()；count 值减一；c==capacity;现在队列有一个空的位置，所以调用 signalNotFull()激活 notFull 条件队列里面等待最久的一 个线程。 LinkedBlockingQueue 简单示例 并发库中的 BlockingQueue 是一个比较好玩的类，顾名思义，就是阻塞队列。该类主要提供了两个方法 put()和 take()，前者将一个对象放到队列中，如果队列已经满了，就等待直到有空闲节点；后者从 head 取一个对象，如果没 有对象，就等待直到有可取的对象。 下面的例子比较简单，一个读线程，用于将要处理的文件对象添加到阻塞队列中， 另外四个写线程用于取出文件 对象，为了模拟写操作耗时长的特点，特让线程睡眠一段随机长度的时间。另外，该 Demo 也使用到了线程池和原子 整型 （AtomicInteger），AtomicInteger 可以在并发情况下达到原子化更新，避免使用了 synchronized，而且性能 非常高。由 于阻塞队列的 put 和 take 操作会阻塞，为了使线程退出，特在队列中添加了一个“标识”，算法中也叫 “哨兵”，当发现这个哨兵后，写线程就退出。 当然线程池也要显式退出了。 package concurrent; import java.io.File; import java.io.FileFilter; import java.util.concurrent.BlockingQueue; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.atomic.AtomicInteger; public class TestBlockingQueue { static long randomTime() { return (long) (Math.random() * 1000); } 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 164 public static void main(String[] args) { // 能容纳 100 个文件 final BlockingQueue queue = new LinkedBlockingQueue(100); // 线程池 final ExecutorService exec = Executors.newFixedThreadPool(5); final File root = new File("F:\\JavaLib"); // 完成标志 final File exitFile = new File(""); // 读个数 final AtomicInteger rc = new AtomicInteger(); // 写个数 final AtomicInteger wc = new AtomicInteger(); // 读线程 Runnable read = new Runnable() { public void run() { scanFile(root); scanFile(exitFile); } public void scanFile(File file) { if (file.isDirectory()) { File[] files = file.listFiles(new FileFilter() { public boolean accept(File pathname) { return pathname.isDirectory() || pathname.getPath().endsWith(".java"); } }); for (File one : files) scanFile(one); } else { try { int index = rc.incrementAndGet(); System.out.println("Read0: " + index + " " + file.getPath()); queue.put(file); } catch (InterruptedException e) { } } } }; exec.submit(read); // 四个写线程 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 165 for (int index = 0; index < 4; index++) { // write thread final int NO = index; Runnable write = new Runnable() { String threadName = "Write" + NO; public void run() { while (true) { try { Thread.sleep(randomTime()); int index = wc.incrementAndGet(); File file = queue.take(); // 队列已经无对象 if (file == exitFile) { // 再次添加"标志"，以让其他线程正常退出 queue.put(exitFile); break; } System.out.println(threadName + ": " + index + " " + file.getPath()); } catch (InterruptedException e) { } } } }; exec.submit(write); } exec.shutdown(); } } ➢ PriorityBlockingQueue 无界阻塞优先级队列 PriorityBlockingQueue 是带优先级的无界阻塞队列，每次出队都返回优先级最高的元素，是二叉树最小堆的实 现，研究过数组方式存放最小堆节点的都知道，直接遍历队列元素是无序的。 PriorityBlockingQueue 类图结构 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 166 如图 PriorityBlockingQueue 内部有个数组 queue 用来存放队列元素，size 用来存放队列元素个数， allocationSpinLockOffset 是用来在扩容队列时候做 cas 的，目的是保证只有一个线程可以进行扩容。 由于这是一个优先级队列所以有个比较器 comparator 用来比较元素大小。lock 独占锁对象用来控制同时只能有 一个线程可以进行入队出队操作。notEmpty 条件变量用来实现 take 方法阻塞模式。这里没有 notFull 条件变量是因 为这里的 put 操作是非阻塞的，为啥要设计为非阻塞的是因为这是无界队列。 最后 PriorityQueue q 用来搞序列化的。 如下构造函数，默认队列容量为 11，默认比较器为 null; private static final int DEFAULT_INITIAL_CAPACITY = 11; public PriorityBlockingQueue() { this(DEFAULT_INITIAL_CAPACITY, null); } 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 167 public PriorityBlockingQueue(int initialCapacity) { this(initialCapacity, null); } public PriorityBlockingQueue(int initialCapacity, Comparator comparator) { if (initialCapacity < 1) throw new IllegalArgumentException(); this.lock = new ReentrantLock(); this.notEmpty = lock.newCondition(); this.comparator = comparator; this.queue = new Object[initialCapacity]; } PriorityBlockingQueue 方法 ✓ Offer 操作 在队列插入一个元素，由于是无界队列，所以一直为成功返回 true; public boolean offer(E e) { if (e == null) throw new NullPointerException(); final ReentrantLock lock = this.lock; lock.lock(); int n, cap; Object[] array; //如果当前元素个数>=队列容量，则扩容(1) while ((n = size) >= (cap = (array = queue).length)) tryGrow(array, cap); try { Comparator cmp = comparator; //默认比较器为 null if (cmp == null)(2) siftUpComparable(n, e, array); else //自定义比较器(3) 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 168 siftUpUsingComparator(n, e, array, cmp); //队列元素增加 1，并且激活 notEmpty 的条件队列里面的一个阻塞线程 size = n + 1;（9） notEmpty.signal(); } finally { lock.unlock(); } return true; } 主流程比较简单，下面看看两个主要函数 private void tryGrow(Object[] array, int oldCap) { lock.unlock(); //must release and then re-acquire main lock Object[] newArray = null; //cas 成功则扩容(4) if (allocationSpinLock == 0 && UNSAFE.compareAndSwapInt(this, allocationSpinLockOffset, 0, 1)) { try { //oldGap<64 则扩容新增 oldcap+2,否者扩容 50%，并且最大为 MAX_ARRAY_SIZE int newCap = oldCap + ((oldCap < 64) ? (oldCap + 2) : // grow faster if small (oldCap >> 1)); if (newCap - MAX_ARRAY_SIZE > 0) { // possible overflow int minCap = oldCap + 1; if (minCap < 0 || minCap > MAX_ARRAY_SIZE) throw new OutOfMemoryError(); newCap = MAX_ARRAY_SIZE; } if (newCap > oldCap && queue == array) newArray = new Object[newCap]; } finally { allocationSpinLock = 0; } } //第一个线程 cas 成功后，第二个线程会进入这个地方，然后第二个线程让出 cpu，尽量让第一个线程执行下面点获取锁，但 是这得不到肯定的保证。(5) if (newArray == null) // back off if another thread is allocating Thread.yield(); lock.lock();(6) 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 169 if (newArray != null && queue == array) { queue = newArray; System.arraycopy(array, 0, newArray, 0, oldCap); } } tryGrow 目的是扩容，这里要思考下为啥在扩容前要先释放锁，然后使用 cas 控制只有一个线程可以扩容成功。 我的理解是为了性能，因为扩容时候是需要花时间的，如果这些操作时候还占用锁那么其他线程在这个时候是不能进 行出队操作的，也不能进行入队操作，这大大降低了并发性。 所以在扩容前释放锁，这允许其他出队线程可以进行出队操作，但是由于释放了锁，所以也允许在扩容时候进行 入队操作，这就会导致多个线程进行扩容会出现问题，所以这里使用了一个 spinlock 用 cas 控制只有一个线程可以进 行扩容，失败的线程调用 Thread.yield()让出 cpu，目的意在让扩容线程扩容后优先调用 lock.lock 重新获取锁，但是 这得不到一定的保证，有可能调用 Thread.yield()的线程先获取了锁。 那 copy 元素数据到新数组为啥放到获取锁后面那?原因应该是因为可见性问题，因为 queue 并没有被 volatile 修 饰。另外有可能在扩容时候进行了出队操作，如果直接拷贝可能看到的数组元素不是最新的。而通过调用 Lock 后，获 取的数组则是最新的，并且在释放锁前 数组内容不会变化。 具体建堆算法： private static void siftUpComparable(int k, T x, Object[] array) { Comparable key = (Comparable) x; //队列元素个数>0 则判断插入位置，否者直接入队(7) while (k > 0) { int parent = (k - 1) >>> 1; Object e = array[parent]; if (key.compareTo((T) e) >= 0) break; array[k] = e; k = parent; } array[k] = key;(8) } 下面用图说话模拟下过程： 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 170 假设队列容量为 2 • 第一次 offer(2)时候 执行（1)为 false 所以执行（2），由于 k=n=size=0;所以执行（8）元素入队，然执行（9）size+1; 现在队列状态: • 第二次 offer(4)时候 执行（1)为 false，所以执行（2）由于 k=1,所以进入 while 循环，parent=0;e=2;key=4;key>e 所以 break; 然后把 4 存到数据下标为 1 的地方，这时候队列状态为： • 第三次 offer(4)时候 执行（1)为 true,所以调用 tryGrow,由于 2<64 所以 newCap=2 + (2+2)=6;然后创建新数组并拷贝，然后调 用 siftUpComparable；k=2>0 进入循环 parent=0;e=2;key=6;key>e 所以 break;然后把 6 放入下标为 2 的地方，现在队列状态： 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 171 • 第四次 offer(1)时候 执行（1)为 false，所以执行（2）由于 k=3,所以进入 while 循环，parent=0;e=2;key=1; key cmp = comparator; if (cmp == null)//cmp=null 则调用这个，把对尾元素位置插入到 0 位置，并且调整堆为最小堆(3) siftDownComparable(0, x, array, n); else siftDownUsingComparator(0, x, array, n, cmp); size = n;（4） return result; } } private static void siftDownComparable(int k, T x, Object[] array, int n) { if (n > 0) { Comparable key = (Comparable)x; int half = n >>> 1; // loop while a non-leaf while (k < half) { int child = (k << 1) + 1; // assume left child is least Object c = array[child];（5） int right = child + 1;（6) if (right < n && ((Comparable) c).compareTo((T) array[right]) > 0)(7) c = array[child = right]; if (key.compareTo((T) c) <= 0)(8) break; array[k] = c; k = child; } array[k] = key;(9) } } 下面用图说话模拟下过程： • 第一次调用 poll() 首先执行（1） result=1；然后执行（2）x=2;这时候队列状态 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 173 然后执行（3）后状态为： 执行（4）后的结果： 下面重点说说 siftDownComparable 这个屌屌的建立最小堆的算法： 首先说下思想，其中 k 一开始为 0，x 为数组里面最后一个元素，由于第 0 个元素为树根，被出队时候要被搞掉， 所以建堆要从它的左右孩子节点找一个最小的值来当树根，子树根被搞掉后，会找子树的左右孩子最小的元素来代替， 直到树节点为止，还不明白，没关系，看图说话： 假如当前队列元素： 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 174 那么对于树为： 这时候如果调用了 poll();那么 result=2;x=11;现在树为： 然后看 leftChildVal = 4;rightChildVal = 6; 4<6;所以 c=4;也就是获取根节点的左右孩子值小的那一个； 然后看 11>4 也就是 key>c；然后把 c 放入树根，现在树为： 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 175 然后看根的左边孩子 4 为根的子树我们要为这个字树找一个根节点。 看 leftChildVal = 8;rightChildVal = 10; 8<10;所以 c=8;也就是获取根节点的左右孩子值小的那一个； 然后看 11>8 也就是 key>c；然后把 c 放入树根，现在树为： 这时候 k=3;half=3 所以推出循环，执行（9）后结果为： 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 176 这时候队列为： ✓ Put 操作 内部调用的 offer,由于是无界队列，所以不需要阻塞 public void put(E e) { offer(e); // never need to block } ✓ Take 操作 获取队列头元素，如果队列为空则阻塞。 public E take() throws InterruptedException { final ReentrantLock lock = this.lock; lock.lockInterruptibly(); E result; try { //如果队列为空，则阻塞，把当前线程放入 notEmpty 的条件队列 while ( (result = dequeue()) == null) notEmpty.await(); } finally { lock.unlock(); } 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 177 return result; } 这里是阻塞实现，阻塞后直到入队操作调用 notEmpty.signal 才会返回。 ✓ Size 操作 获取队列元个数，由于加了独占锁所以返回结果是精确的 public int size() { final ReentrantLock lock = this.lock; lock.lock(); try { return size; } finally { lock.unlock(); } } PriorityBlockingQueue 小结 PriorityBlockingQueue 类似于 ArrayBlockingQueue 内部使用一个独占锁来控制同时只有一个线程可以进行入 队和出队，另外前者只使用了一个 notEmpty 条件变量而没有 notFull 这是因为前者是无界队列，当 put 时候永远不 会处于 await 所以也不需要被唤醒。 PriorityBlockingQueue 始终保证出队的元素是优先级最高的元素，并且可以定制优先级的规则，内部通过使用 一个二叉树最小堆算法来维护内部数组，这个数组是可扩容的，当当前元素个数>=最大容量时候会通过算法扩容。 值得注意的是为了避免在扩容操作时候其他线程不能进行出队操作，实现上使用了先释放锁，然后通过 cas 保证 同时只有一个线程可以扩容成功。 PriorityBlockingQueue 示例 PriorityBlockingQueue 类是 JDK 提供的优先级队列 本身是线程安全的 内部使用显示锁 保证线程安全。 PriorityBlockingQueue 存储的对象必须是实现 Comparable 接口的 因为 PriorityBlockingQueue 队列会根据 内部存储的每一个元素的 compareTo 方法比较每个元素的大小。这样在 take 出来的时候会根据优先级 将优先级最 小的最先取出 。 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 178 下面是示例代码 public static PriorityBlockingQueue queue = new PriorityBlockingQueue(); public static void main(String[] args) { queue.add(new User(1,"wu")); queue.add(new User(5,"wu5")); queue.add(new User(23,"wu23")); queue.add(new User(55,"wu55")); queue.add(new User(9,"wu9")); queue.add(new User(3,"wu3")); for (User user : queue) { try { System.out.println(queue.take().name); } catch (InterruptedException e) { e.printStackTrace(); } } } //静态内部类 static class User implements Comparable{ public User(int age,String name) { this.age = age; this.name = name; } int age; String name; @Override public int compareTo(User o) { return this.age > o.age ? -1 : 1; } } ➢ SychronousQueue 同步队列 SynchronousQueue 是一个比较特别的队列，由于在线程池方面有所应用，为了更好的理解线程池的实现原理， 此队列源码中充斥着大量的 CAS 语句，理解起来是有些难度的，为了方便日后回顾，本篇文章会以简洁的图形化方式 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 179 展示该队列底层的实现原理。 SychronousQueue 简单实用 经典的生产者-消费者模式，操作流程是这样的： 有多个生产者，可以并发生产产品，把产品置入队列中，如果队列满了，生产者就会阻塞； 有多个消费者，并发从队列中获取产品，如果队列空了，消费者就会阻塞； 如下面的示意图所示： 生产线程 生产线程 产品1 产品2 产品3 容器 消费线程 消费线程 SynchronousQueue 也是一个队列来的，但它的特别之处在于它内部没有容器，一个生产线程，当它生产产品（即 put 的时候），如果当前没有人想要消费产品(即当前没有线程执行 take)，此生产线程必须阻塞，等待一个消费线程调 用 take 操作，take 操作将会唤醒该生产线程，同时消费线程会获取生产线程的产品（即数据传递），这样的一个过程 称为一次配对过程(当然也可以先 take 后 put,原理是一样的)。 我们用一个简单的代码来验证一下，如下所示： package com.concurrent; import java.util.concurrent.SynchronousQueue; public class SynchronousQueueDemo { public static void main(String[] args) throws InterruptedException { final SynchronousQueue queue = new SynchronousQueue(); 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 180 Thread putThread = new Thread(new Runnable() { @Override public void run() { System.out.println("put thread start"); try { queue.put(1); } catch (InterruptedException e) { } System.out.println("put thread end"); } }); Thread takeThread = new Thread(new Runnable() { @Override public void run() { System.out.println("take thread start"); try { System.out.println("take from putThread: " + queue.take()); } catch (InterruptedException e) { } System.out.println("take thread end"); } }); putThread.start(); Thread.sleep(1000); takeThread.start(); } } 一种输出结果如下： put thread start take thread start take from putThread: 1 put thread end take thread end 从结果可以看出，put 线程执行 queue.put(1) 后就被阻塞了，只有 take 线程进行了消费，put 线程才可以返回。 可以认为这是一种线程与线程间一对一传递消息的模型。 SychronousQueue 实现原理 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 181 不像 ArrayBlockingQueue 、 LinkedBlockingDeque 之 类 的 阻 塞 队 列 依 赖 AQS 实 现 并 发 操 作 ， SynchronousQueue 直接使用 CAS 实现线程的安全访问。 队列的实现策略通常分为公平模式和非公平模式，接下来将分别进行说明。 公平模式下的模型： 公平模式下，底层实现使用的是 TransferQueue 这个内部队列，它有一个 head 和 tail 指针，用于指向当前正在 等待匹配的线程节点。 初始化时，TransferQueue 的状态如下： 接着我们进行一些操作： 1、线程 put1 执行 put(1)操作，由于当前没有配对的消费线程，所以 put1 线程入队列，自旋一小会后睡眠 等待，这时队列状态如下： 2、接着，线程 put2 执行了 put(2)操作，跟前面一样，put2 线程入队列，自旋一小会后睡眠等待，这时队列 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 182 状态如下： 3、这时候，来了一个线程 take1，执行了 take 操作，由于 tail 指向 put2 线程，put2 线程跟 take1 线程配 对了(一 put 一 take)，这时 take1 线程不需要入队，但是请注意了，这时候，要唤醒的线程并不是 put2，而 是 put1。 为何？ 大家应该知道我们现在讲的是公平策略，所谓公平就是谁先入队了，谁就优先被唤醒，我们的例子明显是 put1 应该优先被唤醒。至于读者可能会有一个疑问，明明是 take1 线程跟 put2 线程匹配上了，结果是 put1 线程被 唤醒消费，怎么确保 take1 线程一定可以和次首节点(head.next)也是匹配的呢？其实大家可以拿个纸画一画，就会发 现真的就是这样的。 公平策略总结下来就是：队尾匹配队头出队。 执行后 put1 线程被唤醒，take1 线程的 take()方法返回了 1(put1 线程的数据)，这样就实现了线程间的一对一通 信，这时候内部状态如下： 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 183 4、最后，再来一个线程 take2，执行 take 操作，这时候只有 put2 线程在等候，而且两个线程匹配上了，线 程 put2 被唤醒， take2 线程 take 操作返回了 2(线程 put2 的数据)，这时候队列又回到了起点，如下所示： 以上便是公平模式下，SynchronousQueue 的实现模型。总结下来就是：队尾匹配队头出队，先进先出，体现公 平原则。 非公平模式下的模型： 我们还是使用跟公平模式下一样的操作流程，对比两种策略下有何不同。非公平模式底层的实现使用的是 TransferStack，一个栈，实现中用 head 指针指向栈顶，接着我们看看它的实现模型: 1、线程 put1 执行 put(1)操作，由于当前没有配对的消费线程，所以 put1 线程入栈，自旋一小会后睡眠等 待，这时栈状态如下： put1线程head 栈 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 184 2、接着，线程 put2 再次执行了 put(2)操作，跟前面一样，put2 线程入栈，自旋一小会后睡眠等待，这时栈 状态如下： put1线程 栈 put2线程head 3、这时候，来了一个线程 take1，执行了 take 操作，这时候发现栈顶为 put2 线程，匹配成功，但是实现会 先把 take1 线程入栈，然后 take1 线程循环执行匹配 put2 线程逻辑，一旦发现没有并发冲突，就会把栈顶 指针直接指向 put1 线程 步骤一： put1线程 head 栈 put2线程 Take1线程 循环中匹配put2线程 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 185 步骤二： put1线程head 栈 4、最后，再来一个线程 take2，执行 take 操作，这跟步骤 3 的逻辑基本是一致的，take2 线程入栈， 然后在循环中匹配 put1 线程，最终全部匹配完毕，栈变为空，恢复初始状态，如下图所示： 步骤一： 栈 put1线程 Take2线程 循环中匹配put1线程 head 步骤二： 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 186 栈 head Null 可以从上面流程看出，虽然 put1 线程先入栈了，但是却是后匹配，这就是非公平的由来。 SychronousQueue 总结 SynchronousQueue 由于其独有的线程一一配对通信机制，在大部分平常开发中，可能都不太会用到，但线程池 技术中会有所使用，由于内部没有使用 AQS，而是直接使用 CAS，所以代码理解起来会比较困难，但这并不妨碍我们 理解底层的实现模型，在理解了模型的基础上，有兴趣的话再查阅源码，就会有方向感，看起来也会比较容易，希望本 文有所借鉴意义。 ➢ DeplayQueue 延时无界阻塞队列 在谈到 DelayQueue 的使用和原理的时候，我们首先介绍一下 DelayQueue，DelayQueue 是一个无界阻塞队 列，只有在延迟期满时才能从中提取元素。该队列的头部是延迟期满后保存时间最长的 Delayed 元素。 DelayQueue 阻塞队列在我们系统开发中也常常会用到，例如：缓存系统的设计，缓存中的对象，超过了空闲时 间，需要从缓存中移出；任务调度系统，能够准确的把握任务的执行时间。我们可能需要通过线程处理很多时间上要求 很严格的数据，如果使用普通的线程，我们就需要遍历所有的对象，一个一个的检 查看数据是否过期等，首先这样在 执行上的效率不会太高，其次就是这种设计的风格也大大的影响了数据的精度。一个需要 12:00 点执行的任务可能 12:01 才执行,这样对数据要求很高的系统有更大的弊端。由此我们可以使用 DelayQueue。 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 187 下面将会对 DelayQueue 做一个介绍，然后举个例子。并且提供一个 Delayed 接口的实现和 Sample 代码。 DelayQueue 是一个 BlockingQueue，其特化的参数是 Delayed。（不了解 BlockingQueue 的同学，先去了解 BlockingQueue 再看本文） Delayed 扩展了 Comparable 接口，比较的基准为延时的时间值，Delayed 接口的实现类 getDelay 的返回值应 为固定值（final）。DelayQueue 内部是使用 PriorityQueue 实现的。 DelayQueue = BlockingQueue +PriorityQueue + Delayed DelayQueue 定义和原理 DelayQueue 的关键元素 BlockingQueue、PriorityQueue、Delayed。可以这么说，DelayQueue 是一个使用 优先队列（PriorityQueue）实现的 BlockingQueue，优先队列的比较基准值是时间。 他们的基本定义如下 public interface Comparable { public int compareTo(T o); } public interface Delayed extends Comparable { long getDelay(TimeUnit unit); } public class DelayQueue implements BlockingQueue { private final PriorityQueue q = new PriorityQueue(); } DelayQueue 内部的实现使用了一个优先队列。当调用 DelayQueue 的 offer 方法时，把 Delayed 对象加入到优 先队列 q 中。如下： public boolean offer(E e) { final ReentrantLock lock = this.lock; lock.lock(); try { E first = q.peek(); q.offer(e); if (first == null || e.compareTo(first) < 0) available.signalAll(); return true; } finally { 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 188 lock.unlock(); } } DelayQueue 的 take 方法，把优先队列 q 的 first 拿出来（peek），如果没有达到延时阀值，则进行 await 处理。如下： public E take() throws InterruptedException { final ReentrantLock lock = this.lock; lock.lockInterruptibly(); try { for (;;) { E first = q.peek(); if (first == null) { available.await(); } else { long delay = first.getDelay(TimeUnit.NANOSECONDS); if (delay > 0) { long tl = available.awaitNanos(delay); } else { E x = q.poll(); assert x != null; if (q.size() != 0) available.signalAll(); // wake up other takers return x; } } } } finally { lock.unlock(); } } DelayQueue 实例应用 Ps：为了具有调用行为，存放到 DelayDeque 的元素必须继承 Delayed 接口。Delayed 接口使对象成为延迟对 象，它使存放在 DelayQueue 类中的对象具有了激活日期。该接口强制执行下列两个方法。 一下将使用 Delay 做一个缓存的实现。其中共包括三个类  Pair 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 189  DelayItem  Cache Pair 类： public class Pair { public K first; public V second; public Pair() {} public Pair(K first, V second) { this.first = first; this.second = second; } } 一下是对 Delay 接口的实现： import java.util.concurrent.Delayed; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; public class DelayItem implements Delayed { /** Base of nanosecond timings, to avoid wrapping */ private static final long NANO_ORIGIN = System.nanoTime(); /** * Returns nanosecond time offset by origin */ final static long now() { return System.nanoTime() - NANO_ORIGIN; } /** * Sequence number to break scheduling ties, and in turn to guarantee FIFO order among tied * entries. */ private static final AtomicLong sequencer = new AtomicLong(0); /** Sequence number to break ties FIFO */ private final long sequenceNumber; 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 190 /** The time the task is enabled to execute in nanoTime units */ private final long time; private final T item; public DelayItem(T submit, long timeout) { this.time = now() + timeout; this.item = submit; this.sequenceNumber = sequencer.getAndIncrement(); } public T getItem() { return this.item; } public long getDelay(TimeUnit unit) { long d = unit.convert(time - now(), TimeUnit.NANOSECONDS); return d; } public int compareTo(Delayed other) { if (other == this) // compare zero ONLY if same object return 0; if (other instanceof DelayItem) { DelayItem x = (DelayItem) other; long diff = time - x.time; if (diff < 0) return -1; else if (diff > 0) return 1; else if (sequenceNumber < x.sequenceNumber) return -1; else return 1; } long d = (getDelay(TimeUnit.NANOSECONDS) - other.getDelay(TimeUnit.NANOSECONDS)); return (d == 0) ? 0 : ((d < 0) ? -1 : 1); } } 以下是 Cache 的实现，包括了 put 和 get 方法 import java.util.concurrent.ConcurrentHashMap; 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 191 import java.util.concurrent.ConcurrentMap; import java.util.concurrent.DelayQueue; import java.util.concurrent.TimeUnit; import java.util.logging.Level; import java.util.logging.Logger; public class Cache { private static final Logger LOG = Logger.getLogger(Cache.class.getName()); private ConcurrentMap cacheObjMap = new ConcurrentHashMap(); private DelayQueue>> q = new DelayQueue>>(); private Thread daemonThread; public Cache() { Runnable daemonTask = new Runnable() { public void run() { daemonCheck(); } }; daemonThread = new Thread(daemonTask); daemonThread.setDaemon(true); daemonThread.setName("Cache Daemon"); daemonThread.start(); } private void daemonCheck() { if (LOG.isLoggable(Level.INFO)) LOG.info("cache service started."); for (;;) { try { DelayItem> delayItem = q.take(); if (delayItem != null) { // 超时对象处理 Pair pair = delayItem.getItem(); cacheObjMap.remove(pair.first, pair.second); // compare and remove } } catch (InterruptedException e) { 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 192 if (LOG.isLoggable(Level.SEVERE)) LOG.log(Level.SEVERE, e.getMessage(), e); break; } } if (LOG.isLoggable(Level.INFO)) LOG.info("cache service stopped."); } // 添加缓存对象 public void put(K key, V value, long time, TimeUnit unit) { V oldValue = cacheObjMap.put(key, value); if (oldValue != null) q.remove(key); long nanoTime = TimeUnit.NANOSECONDS.convert(time, unit); q.put(new DelayItem>(new Pair(key, value), nanoTime)); } public V get(K key) { return cacheObjMap.get(key); } 测试 main 方法： // 测试入口函数 public static void main(String[] args) throws Exception { Cache cache = new Cache(); cache.put(1, "aaaa", 3, TimeUnit.SECONDS); Thread.sleep(1000 * 2); { String str = cache.get(1); System.out.println(str); } Thread.sleep(1000 * 2); { String str = cache.get(1); System.out.println(str); } } 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 193 输出结果为： aaaa null 我们看到上面的结果，如果超过延时的时间，那么缓存中数据就会自动丢失，获得就为 null。 C. 并发（Collection）队列-非阻塞队列 ➢ 非阻塞队列 首先我们要简单的理解下什么是非阻塞队列： 与阻塞队列相反，非阻塞队列的执行并不会被阻塞，无论是消费者的出队，还是生产者的入队。 在底层，非阻塞队列使用的是 CAS(compare and swap)来实现线程执行的非阻塞。 非阻塞队列简单操作 与阻塞队列相同，非阻塞队列中的常用方法，也是出队和入队。 入队方法：  add()：底层调用 offer();  offer()：Queue 接口继承下来的方法，实现队列的入队操作，不会阻碍线程的执行，插入成功返回 true； 出队方法：  poll()：移动头结点指针，返回头结点元素，并将头结点元素出队；队列为空，则返回 null；  peek()：移动头结点指针，返回头结点元素，并不会将头结点元素出队；队列为空，则返回 null； ➢ 非阻塞算法 CAS 首先我们需要了解悲观锁和乐观锁 悲观锁：假定并发环境是悲观的，如果发生并发冲突，就会破坏一致性，所以要通过独占锁彻底禁止冲突发 生。有一个经典比喻，“如果你不锁门，那么捣蛋鬼就回闯入并搞得一团糟”，所以“你只能一次打开门放进 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 194 一个人，才能时刻盯紧他”。 乐观锁：假定并发环境是乐观的，即，虽然会有并发冲突，但冲突可发现且不会造成损害，所以，可以不加任 何保护，等发现并发冲突后再决定放弃操作还是重试。可类比的比喻为，“如果你不锁门，那么虽然捣蛋鬼会 闯入，但他们一旦打算破坏你就能知道”，所以“你大可以放进所有人，等发现他们想破坏的时候再做决定”。 通常认为乐观锁的性能比悲观所更高，特别是在某些复杂的场景。这主要由于悲观锁在加锁的同时，也会把 某些不会造成破坏的操作保护起来；而乐观锁的竞争则只发生在最小的并发冲突处，如果用悲观锁来理解， 就是“锁的粒度最小”。但乐观锁的设计往往比较复杂，因此，复杂场景下还是多用悲观锁。 首先保证正确性，有必要的话，再去追求性能。 CAS 乐观锁的实现往往需要硬件的支持，多数处理器都都实现了一个 CAS 指令，实现“Compare And Swap”的 语义（这里的 swap 是“换入”，也就是 set），构成了基本的乐观锁。 CAS 包含 3 个操作数：  需要读写的内存位置 V  进行比较的值 A  拟写入的新值 B 当且仅当位置 V 的值等于 A 时，CAS 才会通过原子方式用新值 B 来更新位置 V 的值；否则不会执行任何操 作。无论位置 V 的值是否等于 A，都将返回 V 原有的值。 一个有意思的事实是，“使用 CAS 控制并发”与“使用乐观锁”并不等价。CAS 只是一种手段，既可以实现 乐观锁，也可以实现悲观锁。乐观、悲观只是一种并发控制的策略。下文将分别用 CAS 实现悲观锁和乐观锁？ ➢ ConcurrentLinkedQueue 非阻塞无界链表队列 ConcurrentLinkedQueue 是一个线程安全的队列，基于链表结构实现，是一个无界队列，理论上来说队列的 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 195 长度可以无限扩大。 与其他队列相同，ConcurrentLinkedQueue 也采用的是先进先出（FIFO）入队规则，对元素进行排序。当 我们向队列中添加元素时，新插入的元素会插入到队列的尾部；而当我们获取一个元素时，它会从队列的头 部中取出。 因为 ConcurrentLinkedQueue 是链表结构，所以当入队时，插入的元素依次向后延伸，形成链表；而出队 时，则从链表的第一个元素开始获取，依次递增； 不知道，我这样形容能否让你对链表的入队、出队产生一个大概的思路！ ConcurrentLinkedQuere 简单示例 值得注意的是，在使用 ConcurrentLinkedQueue 时，如果涉及到队列是否为空的判断，切记不可使用 size()==0 的做法，因为在 size()方法中，是通过遍历整个链表来实现的，在队列元素很多的时候，size()方法十分消耗性能和时 间，只是单纯的判断队列为空使用 isEmpty()即可！！！ public class ConcurrentLinkedQueueTest { public static int threadCount = 10; public static ConcurrentLinkedQueue queue = new ConcurrentLinkedQueue(); static class Offer implements Runnable { public void run() { //不建议使用 queue.size()==0，影响效率。可以使用!queue.isEmpty() if(queue.size()==0){ String ele = new Random().nextInt(Integer.MAX_VALUE)+""; queue.offer(ele); System.out.println("入队元素为"+ele); } } } static class Poll implements Runnable { public void run() { if(!queue.isEmpty()){ String ele = queue.poll(); 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 196 System.out.println("出队元素为"+ele); } } } public static void main(String[] agrs) { ExecutorService executorService = Executors.newFixedThreadPool(4); for(int x=0;x(null); } 如上代码初始化时候会构建一个 item 为 NULL 的空节点作为链表的首尾节点。 ConcurrentLinkedQuere 方法 ✓ Offer 操作 offer 操作是在链表末尾添加一个元素，下面看看实现原理。 public boolean offer(E e) { //e 为 null 则抛出空指针异常 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 198 checkNotNull(e); //构造 Node 节点构造函数内部调用 unsafe.putObject，后面统一讲 final Node newNode = new Node(e); //从尾节点插入 for (Node t = tail, p = t;;) { Node q = p.next; //如果 q=null 说明 p 是尾节点则插入 if (q == null) { //cas 插入（1） if (p.casNext(null, newNode)) { //cas 成功说明新增节点已经被放入链表，然后设置当前尾节点（包含 head，1，3，5.。。个节点为尾节点） if (p != t) // hop two nodes at a time casTail(t, newNode); // Failure is OK. return true; } // Lost CAS race to another thread; re-read next } else if (p == q)//(2) //多线程操作时候，由于 poll 时候会把老的 head 变为自引用，然后 head 的 next 变为新 head，所以这里需要 //重新找新的 head，因为新的 head 后面的节点才是激活的节点 p = (t != (t = tail)) ? t : head; else // 寻找尾节点(3) p = (p != t && t != (t = tail)) ? t : q; } } 从构造函数知道一开始有个 item 为 null 的哨兵节点，并且 head 和 tail 都是指向这个节点，然后当一个线 程调用 offer 时候首先 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 199 如图首先查找尾节点，q==null,p 就是尾节点，所以执行 p.casNext 通过 cas 设置 p 的 next 为新增节点， 这时候 p==t 所以不重新设置尾节点为当前新节点。由于多线程可以调用 offer 方法，所以可能两个线程同时执行到 了（1）进行 cas，那么只有一个会成功（假如线程 1 成功了），成功后的链表为： 失败的线程会循环一次这时候指针为： 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 200 这时候会执行（3）所以 p=q,然后在循环后指针位置为： 所以没有其他线程干扰的情况下会执行（1）执行 cas 把新增节点插入到尾部，没有干扰的情况下线程 2 cas 会成 功，然后去更新尾节点 tail,由于 p!=t 所以更新。这时候链表和指针为： 假如线程 2cas 时候线程 3 也在执行，那么线程 3 会失败，循环一次后，线程 3 的节点状态为： 这时候 p!=t ；并且 t 的原始值为 told，t 的新值为 tnew ,所以 told!=tnew，所以 p=tnew=tail; 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 201 然后在循环一下后节点状态: q==null 所以执行（1）。 现在就差 p==q 这个分支还没有走，这个要在执行 poll 操作后才会出现这个情况。poll 后会存在下面的状 态 这个时候添加元素时候指针分布为： 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 202 所以会执行（2）分支 结果 p=head 然后循环，循环后指针分布： 所以执行(1),然后 p!=t 所以设置 tail 节点。现在分布图： 自引用的节点会被垃圾回收掉。 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 203 ✓ add 操作 add 操作是在链表末尾添加一个元素，下面看看实现原理。 其实内部调用的还是 offer public boolean add(E e) { return offer(e); } ✓ poll 操作 poll 操作是在链表头部获取并且移除一个元素，下面看看实现原理。 public E poll() { restartFromHead: //死循环 for (;;) { //死循环 for (Node h = head, p = h, q;;) { //保存当前节点值 E item = p.item; //当前节点有值则 cas 变为 null（1） if (item != null && p.casItem(item, null)) { //cas 成功标志当前节点以及从链表中移除 if (p != h) // 类似 tail 间隔 2 设置一次头节点（2） updateHead(h, ((q = p.next) != null) ? q : p); return item; } //当前队列为空则返回 null（3） else if ((q = p.next) == null) { updateHead(h, p); return null; } //自引用了，则重新找新的队列头节点（4） else if (p == q) continue restartFromHead; else//(5) p = q; } 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 204 } } final void updateHead(Node h, Node p) { if (h != p && casHead(h, p)) h.lazySetNext(h); } 当队列为空时候： 可知执行（3）这时候有两种情况，第一没有其他线程添加元素时候(3)结果为 true 然后因为 h!=p 为 false 所以直接返回 null。第二在执行 q=p.next 前，其他线程已经添加了一个元素到队列，这时候（3）返回 false，然后 执行（5）p=q,然后循环后节点分布： 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 205 这时候执行（1）分支，进行 cas 把当前节点值值为 null，同时只有一个线程会成功，cas 成功 标示该节点 从队列中移除了，然后 p!=h,调用 updateHead 方法，参数为 h,p;h!=p 所以把 p 变为当前链表 head 节点，然后 h 节点的 next 指向自己。现在状态为： cas 失败 后 会再次循环，这时候分布图为： 这时候执行（3）返回 null. 现在还有个分支（4）没有执行过，那么什么时候会执行那？ 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 206 这时候执行（1）分支，进行 cas 把当前节点值值为 null，同时只有一个线程 A 会成功，cas 成功 标示该 节点从队列中移除了，然后 p!=h,调用 updateHead 方法，假如执行 updateHead 前另外一个线程 B 开始 poll 这时 候它 p 指向为原来的 head 节点，然后当前线程 A 执行 updateHead 这时候 B 线程链表状态为： 所以会执行（4）重新跳到外层循环，获取当前 head,现在状态为： ✓ peek 操作 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 207 peek 操作是获取链表头部一个元素（只读取不移除），下面看看实现原理。 代码与 poll 类似，只是少了 castItem.并且 peek 操作会改变 head 指向，offer 后 head 指向哨兵节点，第 一次 peek 后 head 会指向第一个真的节点元素。 public E peek() { restartFromHead: for (;;) { for (Node h = head, p = h, q;;) { E item = p.item; if (item != null || (q = p.next) == null) { updateHead(h, p); return item; } else if (p == q) continue restartFromHead; else p = q; } } } ✓ size 操作 获取当前队列元素个数，在并发环境下不是很有用，因为使用 CAS 没有加锁所以从调用 size 函数到返回结 果期间有可能增删元素，导致统计的元素个数不精确。 public int size() { int count = 0; for (Node p = first(); p != null; p = succ(p)) if (p.item != null) // 最大返回 Integer.MAX_VALUE if (++count == Integer.MAX_VALUE) break; return count; } //获取第一个队列元素（哨兵元素不算），没有则为 null Node first() { restartFromHead: for (;;) { for (Node h = head, p = h, q;;) { 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 208 boolean hasItem = (p.item != null); if (hasItem || (q = p.next) == null) { updateHead(h, p); return hasItem ? p : null; } else if (p == q) continue restartFromHead; else p = q; } } } //获取当前节点的 next 元素，如果是自引入节点则返回真正头节点 final Node succ(Node p) { Node next = p.next; return (p == next) ? head : next; } ✓ remove 操作 如果队列里面存在该元素则删除给元素，如果存在多个则删除第一个，并返回 true，否者返回 false public boolean remove(Object o) { //查找元素为空，直接返回 false if (o == null) return false; Node pred = null; for (Node p = first(); p != null; p = succ(p)) { E item = p.item; //相等则使用 cas 值 null,同时一个线程成功，失败的线程循环查找队列中其他元素是否有匹配的。 if (item != null && o.equals(item) && p.casItem(item, null)) { //获取 next 元素 Node next = succ(p); //如果有前驱节点，并且 next 不为空则链接前驱节点到 next, if (pred != null && next != null) pred.casNext(p, next); return true; } 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 209 pred = p; } return false; } ✓ contains 操作 判断队列里面是否含有指定对象，由于是遍历整个队列，所以类似 size 不是那么精确，有可能调用该方法 时候元素还在队列里面，但是遍历过程中才把该元素删除了，那么就会返回 false. public boolean contains(Object o) { if (o == null) return false; for (Node p = first(); p != null; p = succ(p)) { E item = p.item; if (item != null && o.equals(item)) return true; } return false; } ConcurrentLinkedQuere 的 offer 方法有意思的问题 offer 中有个 判断 t != (t = tail）假如 t=node1;tail=node2;并且 node1!=node2 那么这个判断是 true 还 是 false 那，答案是 true，这个判断是看当前 t 是不是和 tail 相等，相等则返回 true 否者为 false，但是无论 结果是啥执行后 t 的值都是 tail。 下面从字节码来分析下为啥。 • 一个例子 public static void main(String[] args) { int t = 2; int tail = 3; System.out.println(t != (t = tail)); } 结果为：true • 字节码文件 C:\Users\Simple\Desktop\TeacherCode\Crm_Test\build\classes\com\itheima\crm\util>javap -c Test001 警告: 二进制文件 Test001 包含 com.itheima.crm.util.Test001 Compiled from "Test001.java" public class com.itheima.crm.util.Test001 { 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 210 public com.itheima.crm.util.Test001(); Code: 0: aload_0 1: invokespecial #8 // Method java/lang/Object."":()V 4: return public static void main(java.lang.String[]); Code: 0: iconst_2 1: istore_1 2: iconst_3 3: istore_2 4: getstatic #16 // Field java/lang/System.out:Ljava/io/PrintStream; 7: iload_1 8: iload_2 9: dup 10: istore_1 11: if_icmpeq 18 14: iconst_1 15: goto 19 18: iconst_0 19: invokevirtual #22 // Method java/io/PrintStream.println:(Z)V 22: return } 我们从上面标黄的字节码文件中分析 一开始栈为空： 栈 • 第 0 行指令作用是把值 2 入栈栈顶元素为 2 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 211 栈 2 • 第 1 行指令作用是将栈顶 int 类型值保存到局部变量 t 中 栈 • 第 2 行指令作用是把值 3 入栈栈顶元素为 3 栈 3 • 第 3 行指令作用是将栈顶 int 类型值保存到局部变量 tail 中。 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 212 • 第 4 调用打印命令 • 第 7 行指令作用是把变量 t 中的值入栈 栈 2 • 第 8 行指令作用是把变量 tail 中的值入栈 栈 2 3 • 现在栈里面的元素为 3、2，并且 3 位于栈顶 • 第 9 行指令作用是当前栈顶元素入栈，所以现在栈内容 3，3，2 栈 2 3 3 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 213 • 第 10 行指令作用是把栈顶元素存放到 t，现在栈内容 3，2 栈 2 3 • 第 11 行指令作用是判断栈顶两个元素值，相等则跳转 18。由于现在栈顶严肃为 3，2 不相等所以返回 true. • 第 14 行指令作用是把 1 入栈 • 然后回头分析下!=是双目运算符，应该是首先把左边的操作数入栈，然后在去计算了右侧操作数。 ConcurrentLinkedQuere 总结 ConcurrentLinkedQueue 使用 CAS 非阻塞算法实现使用 CAS 解决了当前节点与 next 节点之间的安全链接和对 当前节点值的赋值。由于使用 CAS 没有使用锁，所以获取 size 的时候有可能进行 offer，poll 或者 remove 操作，导 致获取的元素个数不精确，所以在并发情况下 size 函数不是很有用。另外第一次 peek 或者 first 时候会把 head 指向 第一个真正的队列元素。 下面总结下如何实现线程安全的，可知入队出队函数都是操作 volatile 变量：head，tail。所以要保证队列线程安 全只需要保证对这两个 Node 操作的可见性和原子性，由于 volatile 本身保证可见性，所以只需要看下多线程下如果 保证对着两个变量操作的原子性。 对于 offer 操作是在 tail 后面添加元素，也就是调用 tail.casNext 方法，而这个方法是使用的 CAS 操作，只有一 个线程会成功，然后失败的线程会循环一下，重新获取 tail，然后执行 casNext 方法。对于 poll 也是这样的。 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 214 ➢ ConcurrentHashMap 非阻塞 Hash 集合 ConcurrentHashMap 是 Java 并发包中提供的一个线程安全且高效的 HashMap 实现，ConcurrentHashMap 在并发编程的场景中使用频率非常之高，本文就来分析下 ConcurrentHashMap 的实现原理，并对其实现原理进行分 析。 ConcurrentLinkedQuere 类图 ConcurrentHashMap 是由 Segment 数组结构和 HashEntry 数组结构组成。Segment 是一种可重入锁 ReentrantLock ，在 ConcurrentHashMap 里 扮 演 锁 的 角 色 ， HashEntry 则用于存储键值对数据。一个 ConcurrentHashMap 里包含一个 Segment 数组，Segment 的结构和 HashMap 类似，是一种数组和链表结构， 一 个 Segment 里包含一个 HashEntry 数组，每个 HashEntry 是一个链表结构的元素， 每个 Segment 守护者一个 HashEntry 数组里的元素,当对 HashEntry 数组的数据进行修改时，必须首先获得它对应的 Segment 锁。 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 215 ConcurrentLinkedQuere 实现原理 众所周知，哈希表是中非常高效，复杂度为 O(1)的数据结构，在 Java 开发中，我们最常见到最频繁使用的就是 HashMap 和 HashTable，但是在线程竞争激烈的并发场景中使用都不够合理。 HashMap ：先说 HashMap，HashMap 是线程不安全的，在并发环境下，可能会形成环状链表（扩容时可能造 成，具体原因自行百度 google 或查看源码分析），导致 get 操作时，cpu 空转，所以，在并发环境中使用 HashMap 是非常危险的。 HashTable ： HashTable 和 HashMap 的实现原理几乎一样，差别无非是 1.HashTable 不允许 key 和 value 为 null；2.HashTable 是线程安全的。但是 HashTable 线程安全的策略实现代价却太大了，简单粗暴，get/put 所有相 关操作都是 synchronized 的，这相当于给整个哈希表加了一把大锁，多线程访问时候，只要有一个线程访问或操作该 对象，那其他线程只能阻塞，相当于将所有的操作串行化，在竞争激烈的并发场景中性能就会非常差。如下图 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 216 HashTable 性能差主要是由于所有操作需要竞争同一把锁，而如果容器中有多把锁，每一把锁锁一段数据，这样 在多线程访问时不同段的数据时，就不会存在锁竞争了，这样便可以有效地提高并发效率。这就是 ConcurrentHashMap 所采用的"分段锁"思想 ConcurrentLinkedQuere 源码解析 ConcurrentHashMap 采用了非常精妙的"分段锁"策略，ConcurrentHashMap 的主干是个 Segment 数组。 final Segment[] segments; Segment 继承了 ReentrantLock，所以它就是一种可重入锁（ReentrantLock)。在 ConcurrentHashMap， 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 217 一个 Segment 就是一个子哈希表，Segment 里维护了一个 HashEntry 数组，并发环境下，对于不同 Segment 的数据进行操作是不用考虑锁竞争的。（就按默认的 ConcurrentLeve 为 16 来讲，理论上就允许 16 个线程 并发执行，有木有很酷） 所以，对于同一个Segment 的操作才需考虑线程同步，不同的Segment则无需考虑。Segment 类似于HashMap， 一个 Segment 维护着一个 HashEntry 数组 transient volatile HashEntry[] table; HashEntry是目前我们提到的最小的逻辑处理单元了。一个ConcurrentHashMap维护一个Segment数组， 一个 Segment 维护一个 HashEntry 数组。 static final class HashEntry { final int hash; final K key; volatile V value; volatile HashEntry next; //其他省略 } 我们说 Segment 类似哈希表，那么一些属性就跟我们之前提到的 HashMap 差不离，比如负载因子 loadFactor，比如阈值 threshold 等等，看下 Segment 的构造方法 Segment(float lf, int threshold, HashEntry[] tab) { this.loadFactor = lf;//负载因子 this.threshold = threshold;//阈值 this.table = tab;//主干数组即 HashEntry 数组 } 我们来看下 ConcurrentHashMap 的构造方法 public ConcurrentHashMap(int initialCapacity, float loadFactor, int concurrencyLevel) { if (!(loadFactor > 0) || initialCapacity < 0 || concurrencyLevel <= 0) throw new IllegalArgumentException(); //MAX_SEGMENTS 为 1<<16=65536，也就是最大并发数为 65536 if (concurrencyLevel > MAX_SEGMENTS) concurrencyLevel = MAX_SEGMENTS; //2 的 sshif 次方等于 ssize，例:ssize=16,sshift=4;ssize=32,sshif=5 int sshift = 0; //ssize 为 segments 数组长度，根据 concurrentLevel 计算得出 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 218 int ssize = 1; while (ssize < concurrencyLevel) { ++sshift; ssize <<= 1; } //segmentShift 和 segmentMask 这两个变量在定位 segment 时会用到，后面会详细讲 this.segmentShift = 32 - sshift; this.segmentMask = ssize - 1; if (initialCapacity > MAXIMUM_CAPACITY) initialCapacity = MAXIMUM_CAPACITY; //计算 cap 的大小，即 Segment 中 HashEntry 的数组长度，cap 也一定为 2 的 n 次方. int c = initialCapacity / ssize; if (c * ssize < initialCapacity) ++c; int cap = MIN_SEGMENT_TABLE_CAPACITY; while (cap < c) cap <<= 1; //创建 segments 数组并初始化第一个 Segment，其余的 Segment 延迟初始化 Segment s0 = new Segment(loadFactor, (int)(cap * loadFactor), (HashEntry[])new HashEntry[cap]); Segment[] ss = (Segment[])new Segment[ssize]; UNSAFE.putOrderedObject(ss, SBASE, s0); this.segments = ss; } 初始化方法有三个参数，如果用户不指定则会使用默认值，initialCapacity 为 16，loadFactor 为 0.75（负载因 子，扩容时需要参考），concurrentLevel 为 16。 从上面的代码可以看出来,Segment 数组的大小 ssize 是由 concurrentLevel 来决定的，但是却不一定等于 concurrentLevel，ssize 一定是大于或等于 concurrentLevel 的最小的 2 的次幂。比如：默认情况下 concurrentLevel 是 16，则 ssize 为 16；若 concurrentLevel 为 14，ssize 为 16；若 concurrentLevel 为 17，则 ssize 为 32。为什么 Segment 的数组大小一定是 2 的次幂？其实主要是便于通过按位与的散列算法来定 位 Segment 的 index。 其实，put 方法对 segment 也会有所体现 public V put(K key, V value) { Segment s; 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 219 //concurrentHashMap 不允许 key/value 为空 if (value == null) throw new NullPointerException(); //hash 函数对 key 的 hashCode 重新散列，避免差劲的不合理的 hashcode，保证散列均匀 int hash = hash(key); //返回的 hash 值无符号右移 segmentShift 位与段掩码进行位运算，定位 segment int j = (hash >>> segmentShift) & segmentMask; if ((s = (Segment)UNSAFE.getObject // nonvolatile; recheck (segments, (j << SSHIFT) + SBASE)) == null) // in ensureSegment s = ensureSegment(j); return s.put(key, hash, value, false); } 从源码看出，put 的主要逻辑也就两步： 1.定位 segment 并确保定位的 Segment 已初始化 2.调用 Segment 的 put 方法。 Ps：关于 segmentShift 和 segmentMask segmentShift 和 segmentMask 这两个全局变量的主要作用是用来定位 Segment，int j =(hash >>> segmentShift) & segmentMask。 segmentMask：段掩码，假如 segments 数组长度为 16，则段掩码为 16-1=15；segments 长度为 32，段掩码为 32-1=31。这样得到的所有 bit 位都为 1，可以更好地保证散列的均匀性 segmentShift：2 的 sshift 次方等于 ssize，segmentShift=32-sshift。若 segments 长度为 16， segmentShift=32-4=28;若 segments 长度为 32，segmentShift=32-5=27。而计算得出的 hash 值最大 为32位，无符号右移segmentShift，则意味着只保留高几位（其余位是没用的），然后与段掩码segmentMask 位运算来定位 Segment。 ConcurrentLinkedQuere 方法 ✓ Get 操作 public V get(Object key) { Segment s; HashEntry[] tab; 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 220 int h = hash(key); long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE; //先定位 Segment，再定位 HashEntry if ((s = (Segment)UNSAFE.getObjectVolatile(segments, u)) != null && (tab = s.table) != null) { for (HashEntry e = (HashEntry) UNSAFE.getObjectVolatile (tab, ((long)(((tab.length - 1) & h)) << TSHIFT) + TBASE); e != null; e = e.next) { K k; if ((k = e.key) == key || (e.hash == h && key.equals(k))) return e.value; } } return null; } get 方法无需加锁，由于其中涉及到的共享变量都使用 volatile 修饰，volatile 可以保证内存可见性，所以 不会读取到过期数据。 来看下 concurrentHashMap 代理到 Segment 上的 put 方法，Segment 中的 put 方法是要加锁的。 只不过是锁粒度细了而已。 ✓ Put 操作 final V put(K key, int hash, V value, boolean onlyIfAbsent) { HashEntry node = tryLock() ? null : scanAndLockForPut(key, hash, value);//tryLock 不成功时会遍历定位到的 HashEnry 位置的链表 （遍历主要是为了使 CPU 缓存链表），若找不到，则创建 HashEntry。tryLock 一定次数后（MAX_SCAN_RETRIES 变量决定），则 lock。若遍历过程中，由于其他线程的操作导致链表头结点变化，则需要重新遍历。 V oldValue; try { HashEntry[] tab = table; int index = (tab.length - 1) & hash;//定位 HashEntry，可以看到，这个 hash 值在定位 Segment 时和在 Segment 中定位 HashEntry 都会用到，只不过定位 Segment 时只用到高几位。 HashEntry first = entryAt(tab, index); for (HashEntry e = first;;) { if (e != null) { K k; if ((k = e.key) == key || (e.hash == hash && key.equals(k))) { oldValue = e.value; if (!onlyIfAbsent) { 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 221 e.value = value; ++modCount; } break; } e = e.next; } else { if (node != null) node.setNext(first); else node = new HashEntry(hash, key, value, first); int c = count + 1; //若 c 超出阈值 threshold，需要扩容并 rehash。扩容后的容量是当前容量的 2 倍。这样可以最大程 度避免之前散列好的 entry 重新散列，具体在另一篇文章中有详细分析，不赘述。扩容并 rehash 的这个过程是比较消耗资源的。 if (c > threshold && tab.length < MAXIMUM_CAPACITY) rehash(node); else setEntryAt(tab, index, node); ++modCount; count = c; oldValue = null; break; } } } finally { unlock(); } return oldValue; } ConcurrentLinkedQuere 总结 ConcurrentHashMap 作为一种线程安全且高效的哈希表的解决方案，尤其其中的"分段锁"的方案，相比 HashTable 的全表锁在性能上的提升非常之大。本文对 ConcurrentHashMap 的实现原理进行了详细分析，并解读了 部分源码，希望能帮助到有需要的童鞋。 ➢ ConcurrentSkipListMap 非阻塞 Hash 跳表集合 大家都是知道 TreeMap，它是使用树形结构的方式进行存储数据的线程不安全的 Map 集合（有序的哈希表）， 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 222 并且可以对 Map 中的 Key 进行排序，Key 中存储的数据需要实现 Comparator 接口或使用 CompareAble 接口的子 类来实现排序。 ConcurrentSkipListMap 也是和 TreeMap，它们都是有序的哈希表。但是，它们是有区别的： 第一，它们的线程安全机制不同，TreeMap 是非线程安全的，而 ConcurrentSkipListMap 是线程安全的。 第二，ConcurrentSkipListMap 是通过跳表实现的，而 TreeMap 是通过红黑树实现的。 那现在我们需要知道说明是跳表。 什么是 SkipList Skip list(跳表）是一种可以代替平衡树的数据结构，默认是按照 Key 值升序的。Skip list 让已排序的数据分布在 多层链表中，以 0-1 随机数决定一个数据的向上攀升与否，通过“空间来换取时间”的一个算法，在每个节点中增加 了向前的指针，在插入、删除、查找时可以忽略一些不可能涉及到的结点，从而提高了效率。 从概率上保持数据结构的平衡比显示的保持数据结构平衡要简单的多。对于大多数应用，用 Skip list 要比用树算 法相对简单。由于 Skip list 比较简单，实现起来会比较容易，虽然和平衡树有着相同的时间复杂度(O(logn)),但是 skip list 的常数项会相对小很多。Skip list 在空间上也比较节省。一个节点平均只需要 1.333 个指针（甚至更少）。 下图为 Skip list 结构图（以 7,14,21,32,37,71,85 序列为例） SkipList 性质 (1) 由很多层结构组成，level 是通过一定的概率随机产生的。 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 223 (2) 每一层都是一个有序的链表，默认是升序，也可以根据创建映射时所提供的 Comparator 进行排序，具 体取决于使用的构造方法。 (3) 最底层(Level 1)的链表包含所有元素。 (4) 如果一个元素出现在 Level i 的链表中，则它在 Level i 之下的链表也都会出现。 (5) 每个节点包含两个指针，一个指向同一链表中的下一个元素，一个指向下面一层的元素。 什么是 ConcurrentSkipListMap ConcurrentSkipListMap 提供了一种线程安全的并发访问的排序映射表。内部是 SkipList（跳表）结构实现， 在理论上能够在 O(log(n))时间内完成查找、插入、删除操作。 注意，调用 ConcurrentSkipListMap 的 size 时，由于多个线程可以同时对映射表进行操作，所以映射表需要遍历整个链表才能返回元素个数，这个操作 是个 O(log(n))的操作。 ConcurrentSkipListMap 数据结构 ConcurrentSkipListMap 的数据结构，如下图所示： 说明：可以看到 ConcurrentSkipListMap 的数据结构使用的是跳表，每一个 HeadIndex、Index 结点都会包含 一个对 Node 的引用，同一垂直方向上的 Index、HeadIndex 结点都包含了最底层的 Node 结点的引用。并且层级越 高，该层级的结点（HeadIndex 和 Index）数越少。Node 结点之间使用单链表结构。 ConcurrentSkipListMap 源码分析 ConcurrentSkipListMap 主要用到了 Node 和 Index 两种节点的存储方式，通过 volatile 关键字实现了并发的操 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 224 作 static final class Node { final K key; volatile Object value;//value 值 volatile Node next;//next 引用 …… } static class Index { final Node node; final Index down;//downy 引用 volatile Index right;//右边引用 …… } ✓ ConcurrentSkipListMap 查找操作 通过 SkipList 的方式进行查找操作：（下图以“查找 91”进行说明：） 红色虚线，表示查找的路径，蓝色向右箭头表示 right 引用；黑色向下箭头表示 down 引用； 下面就是源码中的实现（get 方法是通过 doGet 方法来时实现的） public V get(Object key) { return doGet(key); } //doGet 的实现 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 225 private V doGet(Object okey) { Comparable key = comparable(okey); Node bound = null; Index q = head;//把头结点作为当前节点的前驱节点 Index r = q.right;//前驱节点的右节点作为当前节点 Node n; K k; int c; for (;;) {//遍历 Index d; // 依次遍历 right 节点 if (r != null && (n = r.node) != bound && (k = n.key) != null) { if ((c = key.compareTo(k)) > 0) {//由于 key 都是升序排列的，所有当前关键字大于所要 查找的 key 时继续向右遍历 q = r; r = r.right; continue; } else if (c == 0) { //如果找到了相等的 key 节点，则返回该 Node 的 value 如果 value 为空可能是其他并发 delete 导致的，于是通过另一种 //遍历 findNode 的方式再查找 Object v = n.value; return (v != null)? (V)v : getUsingFindNode(key); } else bound = n; } //如果一个链表中 right 没能找到 key 对应的 value，则调整到其 down 的引用处继续查找 if ((d = q.down) != null) { q = d; r = d.right; } else break; } // 如果通过上面的遍历方式，还没能找到 key 对应的 value，再通过 Node.next 的方式进行查找 for (n = q.node.next; n != null; n = n.next) { if ((k = n.key) != null) { if ((c = key.compareTo(k)) == 0) { Object v = n.value; return (v != null)? (V)v : getUsingFindNode(key); } else if (c < 0) break; } } 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 226 return null; } ✓ ConcurrentSkipListMap 删除操作 通过 SkipList 的方式进行删除操作：（下图以“删除 23”进行说明：） 红色虚线，表示查找的路径，蓝色向右箭头表示 right 引用；黑色向下箭头表示 down 引用； 下面就是源码中的实现（remove 方法是通过 doRemove 方法来时实现的） //remove 操作，通过 doRemove 实现，把所有 level 中出现关键字 key 的地方都 delete 掉 public V remove(Object key) { return doRemove(key, null); } final V doRemove(Object okey, Object value) { Comparable key = comparable(okey); for (;;) { Node b = findPredecessor(key);//得到 key 的前驱（就是比 key 小的最大节点） Node n = b.next;//前驱节点的 next 引用 for (;;) {//遍历 if (n == null)//如果 next 引用为空，直接返回 return null; Node f = n.next; if (n != b.next) // 如果两次获得的 b.next 不是相同的 Node，就跳转 到第一层循环重新获得 b 和 n break; Object v = n.value; if (v == null) { // 当 n 被其他线程 delete 的时候，其 value==null， 此时做辅助处理，并重新获取 b 和 n n.helpDelete(b, f); 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 227 break; } if (v == n || b.value == null) // 当其前驱被 delet 的时候直接跳出，重新获取 b 和 n break; int c = key.compareTo(n.key); if (c < 0) return null; if (c > 0) {//当 key 较大时就继续遍历 b = n; n = f; continue; } if (value != null && !value.equals(v)) return null; if (!n.casValue(v, null)) break; if (!n.appendMarker(f) || !b.casNext(n, f))//casNext 方法就是通过比较和设置 b（前驱） 的 next 节点的方式来实现删除操作 findNode(key); // 通过尝试 findNode 的方式继续 find else { findPredecessor(key); // Clean index if (head.right == null) //如果 head 的 right 引用为空，则表示不存在该 level tryReduceLevel(); } return (V)v; } } } ✓ ConcurrentSkipListMap 插入操作 通过 SkipList 的方式进行插入操作：（下图以“添加 55”的两种情况，进行说明：） 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 228 在 level=2（该 level 存在）的情况下添加 55 的图示:只需在 level<=2 的合适位置插入 55 即可 在 level=4(该 level 不存在，图示 level4 是新建的)的情况下添加 55 的情况：首先新建 level4,然后在 level<=4 的合适位置插入 55。 下面就是源码中的实现（put 方法是通过 doPut 方法来时实现的） //put 操作，通过 doPut 实现 public V put(K key, V value) { if (value == null) throw new NullPointerException(); return doPut(key, value, false); } private V doPut(K kkey, V value, boolean onlyIfAbsent) { Comparable key = comparable(kkey); for (;;) { Node b = findPredecessor(key);//前驱 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 229 Node n = b.next; //定位的过程就是和 get 操作相似 for (;;) { if (n != null) { Node f = n.next; if (n != b.next) // 前后值不一致的情况下，跳转到第一层循环重新获得 b 和 n break;; Object v = n.value; if (v == null) { // n 被 delete 的情况下 n.helpDelete(b, f); break; } if (v == n || b.value == null) // b 被 delete 的情况，重新获取 b 和 n break; int c = key.compareTo(n.key); if (c > 0) { b = n; n = f; continue; } if (c == 0) { if (onlyIfAbsent || n.casValue(v, value)) return (V)v; else break; // restart if lost race to replace value } // else c < 0; fall through } Node z = new Node(kkey, value, n); if (!b.casNext(n, z)) break; // restart if lost race to append to b int level = randomLevel();//得到一个随机的 level 作为该 key-value 插入的最高 level if (level > 0) insertIndex(z, level);//进行插入操作 return null; } } } /** * 获得一个随机的 level 值 */ private int randomLevel() { 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 230 int x = randomSeed; x ^= x << 13; x ^= x >>> 17; randomSeed = x ^= x << 5; if ((x & 0x8001) != 0) // test highest and lowest bits return 0; int level = 1; while (((x >>>= 1) & 1) != 0) ++level; return level; } //执行插入操作：如上图所示，有两种可能的情况： //1.当 level 存在时，对 level<=n 都执行 insert 操作 //2.当 level 不存在（大于目前的最大 level）时，首先添加新的 level，然后在执行操作 1 private void insertIndex(Node z, int level) { HeadIndex h = head; int max = h.level; if (level <= max) {//情况 1 Index idx = null; for (int i = 1; i <= level; ++i)//首先得到一个包含 1~level 个级别的 down 关系的链表， 最后的 inx 为最高 level idx = new Index(z, idx, null); addIndex(idx, h, level);//把最高 level 的 idx 传给 addIndex 方法 } else { // 情况 2 增加一个新的级别 level = max + 1; Index[] idxs = (Index[])new Index[level+1]; Index idx = null; for (int i = 1; i <= level; ++i)//该步骤和情况 1 类似 idxs[i] = idx = new Index(z, idx, null); HeadIndex oldh; int k; for (;;) { oldh = head; int oldLevel = oldh.level; if (level <= oldLevel) { // lost race to add level k = level; break; } HeadIndex newh = oldh; Node oldbase = oldh.node; for (int j = oldLevel+1; j <= level; ++j) newh = new HeadIndex(oldbase, newh, idxs[j], j);//创建新的 if (casHead(oldh, newh)) { k = oldLevel; 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 231 break; } } addIndex(idxs[k], oldh, k); } } /** *在 1~indexlevel 层中插入数据 */ private void addIndex(Index idx, HeadIndex h, int indexLevel) { // insertionLevel 代表要插入的 level，该值会在 indexLevel~1 间遍历一遍 int insertionLevel = indexLevel; Comparable key = comparable(idx.node.key); if (key == null) throw new NullPointerException(); // 和 get 操作类似，不同的就是查找的同时在各个 level 上加入了对应的 key for (;;) { int j = h.level; Index q = h; Index r = q.right; Index t = idx; for (;;) { if (r != null) { Node n = r.node; // compare before deletion check avoids needing recheck int c = key.compareTo(n.key); if (n.value == null) { if (!q.unlink(r)) break; r = q.right; continue; } if (c > 0) { q = r; r = r.right; continue; } } if (j == insertionLevel) {//在该层 level 中执行插入操作 // Don't insert index if node already deleted if (t.indexesDeletedNode()) { findNode(key); // cleans up return; } 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 232 if (!q.link(r, t))//执行 link 操作，其实就是 inset 的实现部分 break; // restart if (--insertionLevel == 0) { // need final deletion check before return if (t.indexesDeletedNode()) findNode(key); return; } } if (--j >= insertionLevel && j < indexLevel)//key 移动到下一层 level t = t.down; q = q.down; r = q.right; } } } ConcurrentLinkedQuere 示例 下面我们看下面示例输出的结果 import java.util.*;import java.util.concurrent.*; /* * ConcurrentSkipListMap 是“线程安全”的哈希表，而 TreeMap 是非线程安全的。 * * 下面是“多个线程同时操作并且遍历 map”的示例 * (01) 当 map 是 ConcurrentSkipListMap 对象时，程序能正常运行。 * (02) 当 map 是 TreeMap 对象时，程序会产生 ConcurrentModificationException 异常。 * */public class ConcurrentSkipListMapDemo1 { // TODO: map 是 TreeMap 对象时，程序会出错。 //private static Map map = new TreeMap(); private static Map map = new ConcurrentSkipListMap(); public static void main(String[] args) { // 同时启动两个线程对 map 进行操作！ new MyThread("a").start(); new MyThread("b").start(); } private static void printAll() { String key, value; Iterator iter = map.entrySet().iterator(); 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 233 while(iter.hasNext()) { Map.Entry entry = (Map.Entry)iter.next(); key = (String)entry.getKey(); value = (String)entry.getValue(); System.out.print("("+key+", "+value+"), "); } System.out.println(); } private static class MyThread extends Thread { MyThread(String name) { super(name); } @Override public void run() { int i = 0; while (i++ < 6) { // “线程名” + "序号" String val = Thread.currentThread().getName()+i; map.put(val, "0"); // 通过“Iterator”遍历 map。 printAll(); } } } } 某一次的运行结果： (a1, 0), (a1, 0), (b1, 0), (b1, 0), (a1, 0), (b1, 0), (b2, 0), (a1, 0), (a1, 0), (a2, 0), (a2, 0), (b1, 0), (b1, 0), (b2, 0), (b2, 0), (b3, 0), (b3, 0), (a1, 0), (a2, 0), (a3, 0), (a1, 0), (b1, 0), (a2, 0), (b2, 0), (a3, 0), (b3, 0), (b1, 0), (b4, 0), (b2, 0), (a1, 0), (b3, 0), (a2, 0), (b4, 0), (a3, 0), (a1, 0), (a4, 0), (a2, 0), (b1, 0), (a3, 0), (b2, 0), (a4, 0), (b3, 0), (b1, 0), (b4, 0), (b2, 0), (b5, 0), (b3, 0), (a1, 0), (b4, 0), (a2, 0), (b5, 0), (a3, 0), (a1, 0), (a4, 0), (a2, 0), (a5, 0), (a3, 0), (b1, 0), (a4, 0), (b2, 0), (a5, 0), (b3, 0), (b1, 0), (b4, 0), (b2, 0), (b5, 0), (b3, 0), (b6, 0), (b4, 0), (a1, 0), (b5, 0), (a2, 0), (b6, 0), (a3, 0), (a4, 0), (a5, 0), (a6, 0), (b1, 0), (b2, 0), (b3, 0), (b4, 0), (b5, 0), (b6, 0), 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 234 结果说明： 示例程序中，启动两个线程(线程 a 和线程 b)分别对 ConcurrentSkipListMap 进行操作。以线程 a 而言，它会先 获取“线程名”+“序号”，然后将该字符串作为 key，将“0”作为 value，插入到 ConcurrentSkipListMap 中；接 着，遍历并输出 ConcurrentSkipListMap 中的全部元素。 线程 b 的操作和线程 a 一样，只不过线程 b 的名字和线程 a 的名字不同。 当 map 是 ConcurrentSkipListMap 对象时，程序能正常运行。如果将 map 改为 TreeMap 时，程序会产生 ConcurrentModificationException 异常。 2) java.util.concurrent.atomic 包 ➢ AtomicBoolean 原子性布尔 AtomicBoolean 是 java.util.concurrent.atomic 包下的原子变量，这个包里面提供了一组原子类。其基本的特性 就是在多线程环境下，当有多个线程同时执行这些类的实例包含的方法时，具有排他性，即当某个线程进入方法，执行 其中的指令时，不会被其他线程打断，而别的线程就像自旋锁一样，一直等到该方法执行完成，才由 JVM 从等待队列 中选择一个另一个线程进入，这只是一种逻辑上的理解。实际上是借助硬件的相关指令来实现的，不会阻塞线程(或者 说只是在硬件级别上阻塞了)。 AtomicBoolean，在这个 Boolean 值的变化的时候不允许在之间插入，保持操作的原子性。 下面将解释重点方法并举例： boolean compareAndSet(expectedValue, updateValue)，这个方法主要两个作用: 1. 比较 AtomicBoolean 和 expect 的值，如果一致，执行方法内的语句。其实就是一个 if 语句 2. 把 AtomicBoolean 的值设成 update,比较最要的是这两件事是一气呵成的，这连个动作之间不会被 打断，任何内部或者外部的语句都不可能在两个动作之间运行。为多线程的控制提供了解决的方案 下面我们从代码上解释： 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 235 首先我们看下在不使用 AtomicBoolean 情况下，代码的运行情况： package zmx.atomic.test; import java.util.concurrent.TimeUnit; public class BarWorker implements Runnable { //静态变量 private static boolean exists = false; private String name; public BarWorker(String name) { this.name = name; } @Override public void run() { if (!exists) { try { TimeUnit.SECONDS.sleep(1); } catch (InterruptedException e1) { // do nothing } exists = true; System.out.println(name + " enter"); try { System.out.println(name + " working"); TimeUnit.SECONDS.sleep(2); } catch (InterruptedException e) { // do nothing } System.out.println(name + " leave"); exists = false; } else { System.out.println(name + " give up"); } } public static void main(String[] args) { 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 236 BarWorker bar1 = new BarWorker("bar1"); BarWorker bar2 = new BarWorker("bar2"); new Thread(bar1).start(); new Thread(bar2).start(); } } 运行结果： bar1 enter bar2 enter bar1 working bar2 working bar1 leave bar2 leave 从上面的运行结果我们可看到，两个线程运行时，都对静态变量 exists 同时做操作，并没有保证 exists 静态变量 的原子性，也就是一个线程在对静态变量 exists 进行操作到时候，其他线程必须等待或不作为。等待一个线程操作完 后，才能对其进行操作。 下面我们将静态变量使用 AtomicBoolean 来进行操作 package zmx.atomic.test; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; public class BarWorker2 implements Runnable { //静态变量使用 AtomicBoolean 进行操作 private static AtomicBoolean exists = new AtomicBoolean(false); private String name; public BarWorker2(String name) { this.name = name; } @Override public void run() { if (exists.compareAndSet(false, true)) { System.out.println(name + " enter"); 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 237 try { System.out.println(name + " working"); TimeUnit.SECONDS.sleep(2); } catch (InterruptedException e) { // do nothing } System.out.println(name + " leave"); exists.set(false); } else { System.out.println(name + " give up"); } } public static void main(String[] args) { BarWorker2 bar1 = new BarWorker2("bar1"); BarWorker2 bar2 = new BarWorker2("bar2"); new Thread(bar1).start(); new Thread(bar2).start(); } } 运行结果： bar1 enter bar1 working bar2 give up bar1 leave 可以从上面的运行结果看出仅仅一个线程进行工作，因为 exists.compareAndSet(false, true)提供了原子性操作， 比较和赋值操作组成了一个原子操作, 中间不会提供可乘之机。使得一个线程操作，其他线程等待或不作为。 下面我们简单介绍下 AtomicBoolean 的 API 创建一个 AtomicBoolean 你可以这样创建一个 AtomicBoolean： AtomicBoolean atomicBoolean = new AtomicBoolean(); 以上示例新建了一个默认值为 false 的 AtomicBoolean。如果你想要为 AtomicBoolean 实例设置一个显式的 初始值，那么你可以将初始值传给 AtomicBoolean 的构造子： AtomicBoolean atomicBoolean = new AtomicBoolean(true); 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 238 获得 AtomicBoolean 的值 你可以通过使用 get() 方法来获取一个 AtomicBoolean 的值。示例如下： AtomicBoolean atomicBoolean = new AtomicBoolean(true); boolean value = atomicBoolean.get(); 设置 AtomicBoolean 的值 你可以通过使用 set() 方法来设置一个 AtomicBoolean 的值。 示例如下： AtomicBoolean atomicBoolean = new AtomicBoolean(true); atomicBoolean.set(false); 以上代码执行后 AtomicBoolean 的值为 false。 交换 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; 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 239 boolean wasNewValueSet = atomicBoolean.compareAndSet( expectedValue, newValue); 本示例对 AtomicBoolean 的当前值与 true 值进行比较，如果相等，将 AtomicBoolean 的值更新为 false ➢ AtomicInteger 原子性整型 AtomicInteger，一个提供原子操作的 Integer 的类。在 Java 语言中，++i 和 i++操作并不是线程安全的， 在使用的时候，不可避免的会用到 synchronized 关键字。而 AtomicInteger 则通过一种线程安全的加减操 作接口。 我们先来看看 AtomicInteger 给我们提供了什么方法： public final int get() //获取当前的值 public final int getAndSet(int newValue)//获取当前的值，并设置新的值 public final int getAndIncrement()//获取当前的值，并自增 public final int getAndDecrement() //获取当前的值，并自减 public final int getAndAdd(int delta) //获取当前的值，并加上预期的值 下面通过两个简单的例子来看一下 AtomicInteger 的优势在哪: 普通线程同步: class Test2 { private volatile int count = 0; public synchronized void increment() { count++; //若要线程安全执行执行 count++，需要加锁 } public int getCount() { return count; } } 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 240 使用 AtomicInteger: class Test2 { private AtomicInteger count = new AtomicInteger(); public void increment() { count.incrementAndGet(); } //使用 AtomicInteger 之后，不需要加锁，也可以实现线程安全。 public int getCount() { return count.get(); } } 从上面的例子中我们可以看出：使用 AtomicInteger 是非常的安全的.而且因为 AtomicInteger 由硬件提供原子 操作指令实现的。在非激烈竞争的情况下，开销更小，速度更快。AtomicInteger 是使用非阻塞算法来实现并发控制 的。AtomicInteger 的关键域只有一下 3 个： // setup to use Unsafe.compareAndSwapInt for updates private static final Unsafe unsafe = Unsafe.getUnsafe(); private static final long valueOffset; static { try { valueOffset= unsafe.objectFieldOffset(AtomicInteger.class.getDeclaredField("value")); } catch (Exception ex) { throw new Error(ex); } } private volatile int value; 这里， unsafe 是 java 提供的获得对对象内存地址访问的类，注释已经清楚的写出了，它的作用就是在更新操作 时提供“比较并替换”的作用。实际上就是 AtomicInteger 中的一个工具。valueOffset 是用来记录 value 本身在内 存的便宜地址的，这个记录，也主要是为了在更新操作在内存中找到 value 的位置，方便比较。 注意：value 是用来存储整数的时间变量，这里被声明为 volatile，就是为了保证在更新操作时，当前线程可以拿 到 value 最新的值（并发环境下，value 可能已经被其他线程更新了）。 优点:最大的好处就是可以避免多线程的优先级倒置和死锁情况的发生，提升在高并发处理下的性能。 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 241 下面我们简单介绍下 AtomicInteger 的 API 创建一个 AtomicInteger 创建一个 AtomicInteger 示例如下： AtomicInteger atomicInteger = new AtomicInteger(); 本示例将创建一个初始值为 0 的 AtomicInteger。如果你想要创建一个给定初始值的 AtomicInteger，你可 以这样： AtomicInteger atomicInteger = new AtomicInteger(123); 本示例将 123 作为参数传给 AtomicInteger 的构造子，它将设置 AtomicInteger 实例的初始值为 123。 获得 AtomicInteger 的值 你可以使用 get() 方法获取 AtomicInteger 实例的值。示例如下： 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); 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 242 本示例首先新建一个初始值为 123 的 AtomicInteger 实例。然后将 AtomicInteger 与期望值 123 进行比较， 如果相等，将 AtomicInteger 的值更新为 234。 增加 AtomicInteger 的值 AtomicInteger 类包含有一些方法，通过它们你可以增加 AtomicInteger 的值，并获取其值。这些方法如下： public final int addAndGet(int addValue)//在原来的数值上增加新的值，并返回新值 public final int getAndIncrement()//获取当前的值，并自增 public final int incrementAndget() //自减，并获得自减后的值 public final int getAndAdd(int delta) //获取当前的值，并加上预期的值 第一个 addAndGet() 方法给 AtomicInteger 增加了一个值，然后返回增加后的值。getAndAdd() 方法为 AtomicInteger 增加了一个值，但返回的是增加以前的 AtomicInteger 的值。具体使用哪一个取决于你的应用场景。 以下是这两种方法的示例： AtomicInteger atomicInteger = new AtomicInteger(); System.out.println(atomicInteger.getAndAdd(10)); System.out.println(atomicInteger.addAndGet(10)); 本示例将打印出 0 和 20。例子中，第二行拿到的是加 10 之前的 AtomicInteger 的值。加 10 之前的值是 0。 第三行将 AtomicInteger 的值再加 10，并返回加操作之后的值。该值现在是为 20。你当然也可以使用这俩方法为 AtomicInteger 添加负值。结果实际是一个减法操作。getAndIncrement() 和 incrementAndGet() 方法类似于 getAndAdd() 和 addAndGet()，但每次只将 AtomicInteger 的值加 1。 减小 AtomicInteger 的值 AtomicInteger 类还提供了一些减小 AtomicInteger 的值的原子性方法。这些方法是： public final int decrementAndGet() public final int getAndDecrement() decrementAndGet() 将 AtomicInteger 的值减一，并返回减一后的值。getAndDecrement() 也将 AtomicInteger 的值减一，但它返回的是减一之前的值。 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 243 ➢ AtomicIntegerArray 原子性整型数组 java.util.concurrent.atomic.AtomicIntegerArray 类提供了可以以原子方式读取和写入的底层 int 数组的操作， 还包含高级原子操作。 AtomicIntegerArray 支持对底层 int 数组变量的原子操作。 它具有获取和设置方法，如在变 量上的读取和写入。 也就是说，一个集合与同一变量上的任何后续 get 相关联。 原子 compareAndSet 方法也具有 这些内存一致性功能。 AtomicIntegerArray 本质上是对 int[]类型的封装。使用 Unsafe 类通过 CAS 的方式控制 int[]在多线程下的安全 性。它提供了以下几个核心 API： //获得数组第 i 个下标的元素 public final int get(int i) //获得数组的长度 public final int length() //将数组第 i 个下标设置为 newValue，并返回旧的值 public final int getAndSet(int i, int newValue) //进行 CAS 操作，如果第 i 个下标的元素等于 expect，则设置为 update，设置成功返回 true public final boolean compareAndSet(int i, int expect, int update) //将第 i 个下标的元素加 1 public final int getAndIncrement(int i) //将第 i 个下标的元素减 1 public final int getAndDecrement(int i) //将第 i 个下标的元素增加 delta（delta 可以是负数） public final int getAndAdd(int i, int delta) 下面给出一个简单的示例，展示 AtomicIntegerArray 使用： public class AtomicIntegerArrayDemo { static AtomicIntegerArray arr = new AtomicIntegerArray(10); public static class AddThread implements Runnable{ public void run(){ for(int k=0;k<10000;k++) arr.getAndIncrement(k%arr.length()); } } public static void main(String[] args) throws InterruptedException { Thread[] ts=new Thread[10]; for(int k=0;k<10;k++){ ts[k]=new Thread(new AddThread()); } 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 244 for(int k=0;k<10;k++){ts[k].start();} for(int k=0;k<10;k++){ts[k].join();} System.out.println(arr); } } 输出结果： [10000, 10000, 10000, 10000, 10000, 10000, 10000, 10000, 10000, 10000] 上述代码第 2 行，申明了一个内含 10 个元素的数组。第 3 行定义的线程对数组内 10 个元素进行累加操作，每个 元素各加 1000 次。第 11 行，开启 10 个这样的线程。因此，可以预测，如果线程安全，数组内 10 个元素的值必然都 是 10000。反之，如果线程不安全，则部分或者全部数值会小于 10000。 ➢ AtomicLong、AtomicLongArray 原子性整型数组 AtomicLong、AtomicLongArray 的 API 跟 AtomicInteger、AtomicIntegerArray 在使用方法都是差不多的。 区别在于用前者是使用原子方式更新的 long 值和 long 数组，后者是使用原子方式更新的 Integer 值和 Integer 数组。 两者的相同处在于它们此类确实扩展了 Number，允许那些处理基于数字类的工具和实用工具进行统一访问。在实际 开发中，它们分别用于不同的场景。这个就具体情况具体分析了，下面将举例说明 AtomicLong 的使用场景（使用 AtomicLong 生成自增长 ID），其他就不在过多介绍。 import java.util.ArrayList; import java.util.Collections; import java.util.HashSet; import java.util.List; import java.util.Set; import java.util.concurrent.atomic.AtomicLong; public class AtomicLongTest { /** * @param args */ public static void main(String[] args) { final AtomicLong orderIdGenerator = new AtomicLong(0); final List orders = Collections .synchronizedList(new ArrayList()); 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 245 for (int i = 0; i < 10; i++) { Thread orderCreationThread = new Thread(new Runnable() { public void run() { for (int i = 0; i < 10; i++) { long orderId = orderIdGenerator.incrementAndGet(); Item order = new Item(Thread.currentThread().getName(), orderId); orders.add(order); } } }); orderCreationThread.setName("Order Creation Thread " + i); orderCreationThread.start(); } try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } Set orderIds = new HashSet(); for (Item order : orders) { orderIds.add(order.getID()); System.out.println("Order name:" + order.getItemName() +"----"+"Order id:" + order.getID()); } } } class Item { String itemName; long id; Item(String n, long id) { this.itemName = n; this.id = id; } public String getItemName() { return itemName; } public long getID() { 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 246 return id; } } 输出： Order name:Order Creation Thread 0----Order id:1 Order name:Order Creation Thread 1----Order id:2 Order name:Order Creation Thread 0----Order id:4 Order name:Order Creation Thread 1----Order id:5 Order name:Order Creation Thread 3----Order id:3 Order name:Order Creation Thread 0----Order id:7 Order name:Order Creation Thread 1----Order id:6 ........ Order name:Order Creation Thread 2----Order id:100 从运行结果我们看到，不管是哪个线程。它们获得的 ID 是不会重复的，保证的 ID 生成的原子性，避免了线程安 全上的问题。 3) java.util.concurrent.lock 包 待续... （三）多线程面试题 1. 多线程的创建方式（2017-11-23-wzz） （1）、继承 Thread 类：但 Thread 本质上也是实现了 Runnable 接口的一个实例，它代表一个线程的实例，并 且，启动线程的唯一方法就是通过 Thread 类的 start()实例方法。start()方法是一个 native 方法，它将启动一个新线 程，并执行 run()方法。这种方式实现多线程很简单，通过自己的类直接 extend Thread，并复写 run()方法，就可以 启动新线程并执行自己定义的 run()方法。例如：继承 Thread 类实现多线程，并在合适的地方启动线程 1．public class MyThread extends Thread { 2． public void run() { 3． System.out.println("MyThread.run()"); 4． } 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 247 5．} 6．MyThread myThread1 = new MyThread(); 7．MyThread myThread2 = new MyThread(); 8．myThread1.start(); 9．myThread2.start(); （2）、实现 Runnable 接口的方式实现多线程，并且实例化 Thread，传入自己的 Thread 实例，调用 run( )方法 1．public class MyThread implements Runnable { 2． public void run() { 3． System.out.println("MyThread.run()"); 4． } 5．} 6．MyThread myThread = new MyThread(); 7．Thread thread = new Thread(myThread); 8．thread.start(); （3）、使用 ExecutorService、Callable、Future 实现有返回结果的多线程：ExecutorService、Callable、Future 这个对象实际上都是属于 Executor 框架中的功能类。想要详细了解 Executor 框 架 的 可 以 访 问 http://www.javaeye.com/topic/366591 ，这里面对该框架做了很详细的解释。返回结果的线程是在 JDK1.5 中引入 的新特征，确实很实用，有了这种特征我就不需要再为了得到返回值而大费周折了，而且即便实现了也可能漏洞百出。 可返回值的任务必须实现 Callable 接口，类似的，无返回值的任务必须 Runnable 接口。执行 Callable 任务后，可以 获取一个 Future 的对象，在该对象上调用 get 就可以获取到 Callable 任务返回的 Object 了，再结合线程池接口 ExecutorService 就可以实现传说中有返回结果的多线程了。下面提供了一个完整的有返回结果的多线程测试例子，在 JDK1.5 下验证过没问题可以直接使用。代码如下： 1．import java.util.concurrent.*; 2．import java.util.Date; 3．import java.util.List; 4．import java.util.ArrayList; 5． 6．/** 7．* 有返回值的线程 8．*/ 感恩于心，回报于行。 面试宝典系列-Java http://www.itheima.com Copyright© 2018 黑马程序员 248 9．@SuppressWarnings("unchecked") 10．public class Test { 11．public static void main(String[] args) throws ExecutionException, 12． InterruptedException { 13． System.out.println("----程序开始运行----"); 14． Date date1 = new Date(); 15． 16． int taskSize = 5; 17． // 创建一个线程池 18． ExecutorService pool = Executors.newFixedThreadPool(taskSize); 19． // 创建多个有返回值的任务 20． List list = new ArrayList(); 21． for (int i = 0; i < taskSize; i++) { 22． Callable c = new MyCallable(i + " "); 23． // 执行任务并获取 Future 对象 24． Future f = pool.submit(c); 25． // System.out.println(">>>" + f.get().toString()); 26． list.add(f); 27． } 28． // 关闭线程池 29． pool.shutdown(); 30． 31． // 获取所有并发任务的运行结果 32． for (Future f : list) { 33． // 从 Future 对象上获取任务的返回值，并输出到控制台 34． System.out.println(">>>" + f.get().toString()); 35． } 36． 37． Date date2 = new Date(); 38． System.out.println("----程序结束运行----，程序运行时间【" 39． + (date2.getTime() - date1.getTime()) + "毫秒】"); 40．} 41．} 42． 43．class MyCallable implements Callable