并发-Java中的线程池

Java中的线程池

好处：

• 降低资源消耗
• 提高响应速度
• 提高线程的可管理性

线程池的实现原理

• 处理流程
  • 线程池判断核心线程池里是否都在执行任务，如果不是，则创建一个新的工作中线程来执行任务，如果核心线程池里的线程都在执行任务，则进入下一个流程
  • 线程池判断工作中队列是否已经满，如果工作队列没有满，则将新提交的任务存储在这个工作队列里，如果工作队列满了，则进入下个流程
  • 线程池判断线程池的线程是否都处于工作状态，如果没有，则创建一个新的工作线程来执行任务，如果已经满了，则交给饱和策略来处理这个任务

• ThreadPoolExecutor执行execute方法的4种情况

  • 如果当前运行的线程少于corePoolSize，则创建新线程来执行任务（需获取全局锁）
  • 如果运行线程等于或多于corePoolSize,则将任务加入BlockingQueue
  • 如果无法将任务加入BlockingQueue（队列已满），则创建新的线程来处理任务（获取全局锁）
  • 如果创建新线程将使当前运行的线程超出maximumPoolSize，任务将被拒绝，并调用RejectedExcutionHandler.rejectedExecution()方法.

• 工作线程：线程池创建线程时，会将线程封装成工作线程Worker，Worker在执行完任务后，还会循环获取工作队列里的任务来执行

• 线程池中的线程执行任务分两种

  • 在execute()方法中创建一个线程时，会让这个线程执行当前任务
  • 这个线程执行完任务后，会反复从BlockingQueue获取任务来执行

线程池的使用

线程池的创建

new ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler)
1
2
3
4
5
6
7

• corePoolSize： 线程池基本大小，提交一个任务到线程池时，会创建一个线程来执行任务，及时其他空闲的能执行，也要创建，等需要执行的任务数大于线程池基本大小时就不再创建。

  • 如果调用了线程池的prestartAllCoreThreads()方法，线程池会提前创建并启动所有基本线程。

• maximumPoolSize：线程池最大数量，线程池允许创建的最大线程数。如果队列满了，并已创建的线程数小于最大线程数，则线程池会再创建新的线程执行任务。如果使用了无界任务队列这个参数没什么效果。

• keepAliveTime:线程活动保持时间，线程池的工作线程空闲后，保持的存活时间，如果任务很多，并且每个任务执行的时间比较短，可以调大时间 ，提高线程利用率。

• TimeUnit unit:线程活动保持时间的单位，天，小时，分钟，毫秒，微秒，纳秒

• BlockingQueue workQueue：任务队列，用于保存等待执行的任务的阻塞队列

  • ArrayBlockingQueue：是一个基于数组结构的有界阻塞队列，此队列按FIFO原则对元素进行排序
  • LinkedBlockingQueue：一个基于链表结构的阻塞队列，FIFO排序，吞吐量高于ArrayBlockingQueue
  • SynchronousQueueu：一个不存储元素的阻塞队列，每个插入操作必须等到另一个线程调用移除操作，否则插入操作一直处于阻塞状态，吞吐量通常高于LinkedBlockingQueue
  • PriorityBlockingQueue：一个具有优先级的无限阻塞队列

• ThreadFactory threadFactory：用于创建线程的工厂，可以通过线程工厂给每个创建出来的线程设置更有意义的名字。

  • new ThreadFactoryBuilder().setNameFormat("XX-task-%d").build();
    1

• RejectedExecutionHandler handler :饱和策略：当队列和线程池都满了，说明线程池处于饱和状态，采用一种策略处理提交的新任务。默认AbortPolicy

• AbortPolicy：直接抛出异常

• CallerRunsPolicy：只用调用者所在线程来运行任务

• DiscardOldestPolicy：丢弃队列里最近的一个任务，并执行当前任务

• DiscardPolicy：不处理，丢弃掉

• 自定义处理策略

向线程池提交任务

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

  • threadsPool.execute(new Runnable(){
    	@Override
    	public void run(){
    	
    	}
    })
    1
    2
    3
    4
    5
    6

• submit()方法：用于提交需要返回值的任务，线程池会返回一个future类型对象，通过这个对象可以判断任务是否执行成功，

  • 通过future的get方法获取返回值

  • get()方法会阻塞当前线程直到任务完成。

  • get（long timeout，TimeUnit unit）会阻塞当前线程一段时间后立即返回，可能没执行完

  • Future