理解阻塞队列和线程池原理(学习笔记二)_线程池原理 阻塞队列满,采用交由主线程执行,导致阻塞队列中的消息被憋死

阻塞队列

队列介绍

队列是一种先进先出的线性表结构，它的插入操作端为队尾，删除操作端为队头，在队列中插入一个元素称为入队，删除一个元素是出队。

阻塞队列介绍

支持阻塞插入方法：当队列满时，会阻塞向队列插入元素的线程，直到有元素出队
支持阻塞移除方法：当队列为空时，获取元素的线程会等待队列变为非空

阻塞队列常用于生产者消费者模式的场景，为了解决生产者和消费者处理效率不平衡的问题，通过阻塞队列来为生产者和消费者解耦，两者不直接通信，而是通过阻塞队列通信，生产者是向阻塞队列添加元素的线程，消费者是从阻塞队列拿元素的线程，阻塞队列相当于元素存放的容器。

一些状态

• 异常：当队列满时，向队列插入元素，会抛出IllegalStateException（Queuefull）异常，当队列为空时获取数据，会抛出NoSuchElementException异常
• 返回特殊值：插入成功返回true，获取不到元素返回null
• 一直阻塞：阻塞队列满时，如果生产者线程往队列里put元素，队列会一直阻塞生产者线程，直到队列可用或者响应中断退出。当队列空时，如果消费者线程从队列里take元素，队列会阻塞住消费者线程，直到队列不为空。
• 超时退出：当队列满时，生产者线程插入数据会被阻塞一段时间，超时会自动退出

有界阻塞列队：

• ArrayBlockingQueue 数组结构组成（非公平阻塞队列，当队列可用，阻塞的线程会抢夺访问资格，生产者和消费者共用一把锁，需要指定大小）
• LinkedBlockingQueue 链表结构组成（生产者消费者用不同的锁，大小默认Integer.MAX_VALUE）

无界阻塞列队：

• PriorityBlockingQueue 支持优先级排序
• DelayQueue 使用优先级队列实现（使用PriorityQueue实现，元素需要hi先Delayed接口，创建时可以指定多久可以获取，延时期满才能取元素）

缓存系统的设计：可以用DelayQueue保存缓存元素的有效期，使用一个线程循环查询DelayQueue，一旦能从DelayQueue中获取元素时，表示缓存有效期到了。

• LinkedTransferQueue 链表结构组成，有tryTransfer和tranfer方法：
  transfer：如果当前有消费者正在等待接收元素，可以立刻把元素传递给消费者，如果没用则放在队列的tail节点，等该元素被消费才返回。
  trytranfer:试探生产者传入元素是否能直接传递给消费者，如果没用消费者等待则返回false，而transfer必须等元素被消费者消费才返回

• SynchronousQueue 不存储元素的阻塞队列，每一个put操作必须等待一个take操作，否正不能继续添加元素，相当于在双方中间递了一下元素
  LinkedBlockingDeque 链表结构组成的双向阻塞队列，列队两端都可以插入和删除元素，多线程入队时减少了一半竞争

不管有界无界，都会产生阻塞，无界如果消费者拿元素为空，同样阻塞
无界并不是真正意义的无界，会受到系统资源的限制，处理不及时，内存超出界限，会出现OOM。

线程池

1. 使用线程池可以通过重复利用已创建的线程降低线程创建和销毁造成的资源消耗
2. 使用线程可以提高响应速度，省去创建线程和销毁线程的时间
3. 提高线程的可管理性，统一分配，调优和监控

ThreadPoolExecutor相关类的联系

Excutor:是一个接口，是Excutor框架的基础，将任务的提交和执行分开
ExcutorService接口继承了Excutor，做了submit，shutdown扩展，算是真的下线程池接口
AbstractExcutorService抽象类，实现了ExcutorService接口中的大部分方法
ThreadPoolExecutor线程池的核心实现类，用来执行被提交的任务。
ScheduledExecutorService接口继承了ExcutorService接口，提供周期性执行ExcutorService
ScheduledThreadPoolExcutor是一个实现类，可以在确定的延迟后执行命令，或定期执行命令，比Timer更灵活，功能更强大。

ThreadPoolExecutor参数

public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,
long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable>
workQueue,ThreadFactory threadFactory,
RejectedExecutionHandler handler)

1
2
3
4
5
6
7

• corePoolSize核心线程数，prestartAllCoreThreads()方法会提前创建启动核心线程

• maximumPoolSize，最大线程数
  设置技巧
  在创建线程池时，可以先考虑是用来执行什么样的任务
  cpu密集型（计算）：核心线程数 = cpu内核数+1，+1是在使用虚拟内存时最大化使用cpu
  I/O密集型（网络/本地读写数据）：cpu内核数*2
  混合型：前面两种任务数相等最好分开，或者cpu内核数*2

• keepAliveTime，线程空闲时的存活时间，默认只在大于核心线程时生效

• TimeUnit，是keepAliveTime的时间单位

• workQueue，必须是BlockingQueue阻塞队列，当超过核心线程数时候，会进入阻塞队列等待，这样线程池实现了阻塞功能。
  尽量使用有界队列，使用无界会因为无限制往队列添加任务，导致maximumPoolSize，keepAliveTime，TimeUnit都成为无效参数，还有资源耗尽的风险

• threadFactory，创建线程的工厂，通过自定义工厂，可以给新建的线程设置名称，或设置为守护线程等。Executors静态工厂里默认的threadFactory，线程的命名规则是“pool-数字-thread-数字”。

• RejectedExecutionHandler，线程饱和的策略，当队列满了，且没用空闲工作线程时，提供了四种策略：
  bortPolicy：直接抛出异常，默认策略；
  CallerRunsPolicy：用调用者所在的线程来执行任务；
  DiscardOldestPolicy：丢弃阻塞队列中靠最前的任务，并执行当前任务；
  DiscardPolicy：直接丢弃任务；

线程池的工作机制：

1. 如果当前运行的线程少于corePoolSize，则创建新线程来执行任务（注意，执行这一步骤需要获取全局锁）
2. 如果运行的线程等于或多于corePoolSize，则将任务加入BlockingQueue
3. 如果无法将任务加入BlockingQueue（队列已满），则创建新的线程来处理任务
4. 如果创建新线程将使当前运行的线程超出maximumPoolSize，任务将被拒绝，并调用RejectedExecutionHandler.rejectedExecution()方法

提交任务

execute()方法用于提交不需要返回值的任务，所以无法判断任务是否被线程池执行成功。

submit()方法用于提交需要返回值的任务。线程池会返回一个future类型的对象，通过这个future对象可以判断任务是否执行成功，并且可以通过future的get()方法来获取返回值，get()方法会阻塞当前线程直到任务完成，而使用get（long timeout，TimeUnit unit）方法则会阻塞当前线程一段时间后立即返回，这时候有可能任务没有执行完。

关闭线程池
可以通过调用线程池的shutdown或shutdownNow方法关闭线程池。原理是遍历线程池中的工作线程，然后逐个调用线程的interrupt方法来中断线程，所以无法响应中断的任务可能永远无法终止。但是它们存在一定的区别，shutdownNow首先将线程池的状态设置成STOP，然后尝试停止所有的正在执行或暂停任务的线程，并返回等待执行任务的列表，而shutdown只是将线程池的状态设置成SHUTDOWN状态，然后中断所有没有正在执行任务的线程。

只要调用了这两个关闭方法中的任意一个，isShutdown方法就会返回true。当所有的任务都已关闭后，才表示线程池关闭成功，这时调用isTerminaed方法会返回true。至于应该调用哪一种方法来关闭线程池，应该由提交到线程池的任务特性决定，通常调用shutdown方法来关闭线程池，如果任务不一定要执行完，则可以调用shutdownNow方法。

线程池配置策略

首先分析任务的特性：

• 任务的性质：CPU密集型、IO密集型和混合型
• 任务的优先级：高、中和低
• 任务的执行时长：长、中和短
• 任务的依赖性：是否依赖其他系统资源，如数据库连接

CPU密集型：核心线程数 = cpu内核数+1，+1是在使用虚拟内存时最大化使用cpu
IO密集型：cpu内核数*2，IO密集型任务线程并不是一直在执行任务，则应配置尽可能多的线程
混合型：如果可以拆分，将其拆分成一个CPU密集型任务和一个IO密集型任务，只要这两个任务执行的时间相差不是太大，那么分解后执行的吞吐量将高于串行执行的吞吐量。如果这两个任务执行时间相差太大，则没必要进行分解。可以通过Runtime.getRuntime().availableProcessors()方法获得当前设备的CPU个数

优先级不同的任务可以使用优先级列队PriotyBlockingQueue处理。

执行时间不同的任务可以交给不同规模的线程池来处理，或者可以使用优先级队列，让执行时间短的任务先执行。

建议使用有界队列。有界队列能增加系统的稳定性和预警能力，可以根据需要设大一点儿，比如几千。

使用有界队列