大数据处理系列之(一)Java线程池使用

前言:最近在做分布式海量数据处理项目,使用到了java的线程池,所以搜集了一些资料对它的使用做了一下总结和探究,

前面介绍的东西大多都是从网上搜集整理而来。文中最核心的东西在于后面两节无界队列线程池和有界队列线程池的实例

使用以及线上问题处理方案。

1.  为什么要用线程池?

在Java中,如果每当一个请求到达就创建一个新线程,开销是相当大的。在实际使用中,每个请求创建新线程的服务器

在创建和销毁线程上花费的时间和消耗的系统资源,甚至可能要比花在实际处理实际的用户请求的时间和资源要多的多。除

了创建和销毁线程的开销之外,活动的线程也需要消耗系统资源。如果在一个JVM中创建太多的线程,可能会导致系统由于

过度消耗内存或者“切换过度”而导致系统资源不足。为了防止资源不足,服务器应用程序需要一些办法来限制任何给定时刻

处理的请求数目,尽可能减少创建和销毁线程的次数,特别是一些资源耗费比较大的线程的创建和销毁,尽量利用已有对象

来进行服务,这就是“池化资源”技术产生的原因。

线程池主要用来解决线程生命周期开销问题和资源不足问题,通过对多个任务重用线程,线程创建的开销被分摊到多个任

务上了,而且由于在请求到达时线程已经存在,所以消除了创建所带来的延迟。这样,就可以立即请求服务,使应用程序响

应更快。另外,通过适当的调整线程池中的线程数据可以防止出现资源不足的情况。

网上找来的这段话,清晰的描述了为什么要使用线程池,使用线程池有哪些好处。工程项目中使用线程池的场景比比皆是。

本文关注的重点是如何在实战中来使用好线程池这一技术,来满足海量数据大并发用户请求的场景。

2. ThreadPoolExecutor类

Java中的线程池技术主要用的是ThreadPoolExecutor 这个类。先来看这个类的构造函数,

ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,

BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler)

    corePoolSize       线程池维护线程的最少数量

    maximumPoolSize    线程池维护线程的最大数量 

    keepAliveTime      线程池维护线程所允许的空闲时间  

    workQueue          任务队列,用来存放我们所定义的任务处理线程

    threadFactory      线程创建工厂

    handler            线程池对拒绝任务的处理策略

ThreadPoolExecutor 将根据 corePoolSize和 maximumPoolSize 设置的边界自动调整池大小。当新任务在方法

execute(Runnable) 中提交时, 如果运行的线程少于 corePoolSize,则创建新线程来处理请求,即使其他辅助线程是

空闲的。如果运行的线程多于 corePoolSize 而少于 maximumPoolSize,则仅当队列满时才创建新线程。 如果设置的

corePoolSize 和 maximumPoolSize 相同,则创建了固定大小的线程池。

ThreadPoolExecutor是Executors类的实现Executors类里面提供了一些静态工厂,生成一些常用的线程池,主

要有以下几个:

newSingleThreadExecutor:创建一个单线程的线程池。这个线程池只有一个线程在工作,也就是相当于单线程串行执行

所有任务。如果这个唯一的线程因为异常结束,那么会有一个新的线程来替代它。此线程池保证所有任务的执行顺序按照任

务的提交顺序执行。

newFixedThreadPool:创建固定大小的线程池。每次提交一个任务就创建一个线程,直到线程达到线程池的最大大小。线

程池的大小一旦达到最大值就会保持不变,如果某个线程因为执行异常而结束,那么线程池会补充一个新线程。

newCachedThreadPool:创建一个可缓存的线程池。如果线程池的大小超过了处理任务所需要的线程,那么就会回收部分

空闲(60秒不执行任务)的线程,当任务数增加时,此线程池又可以智能的添加新线程来处理任务。此线程池不会对线程池

大小做限制,线程池大小完全依赖于操作系统(或者说JVM)能够创建的最大线程大小。

在实际的项目中,我们会使用得到比较多的是newFixedThreadPool,创建固定大小的线程池,但是这个方法在真实的线上

环境中还是会有很多问题,这个将会在下面一节中详细讲到。

当任务源源不断的过来,而我们的系统又处理不过来的时候,我们要采取的策略是拒绝服务。RejectedExecutionHandler接

口提供了拒绝任务处理的自定义方法的机会。在ThreadPoolExecutor中已经包含四种处理策略。

1)CallerRunsPolicy:线程调用运行该任务的 execute 本身。此策略提供简单的反馈控制机制,能够减缓新任务的提交速度。

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {

if (!e.isShutdown()) {

r.run();

}

}

这个策略显然不想放弃执行任务。但是由于池中已经没有任何资源了,那么就直接使用调用该execute的线程本身来执行。

2)AbortPolicy处理程序遭到拒绝将抛出运行时 RejectedExecutionException

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {

throw new RejectedExecutionException();

}

这种策略直接抛出异常,丢弃任务。

3)DiscardPolicy不能执行的任务将被删除

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {}

这种策略和AbortPolicy几乎一样,也是丢弃任务,只不过他不抛出异常。

4)DiscardOldestPolicy如果执行程序尚未关闭,则位于工作队列头部的任务将被删除,然后重试执行程序(如果再次失败,

则重复此过程)

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {

if (!e.isShutdown()) {

e.getQueue().poll();

e.execute(r);

}

}

该策略就稍微复杂一些,在pool没有关闭的前提下首先丢掉缓存在队列中的最早的任务,然后重新尝试运行该任务。这个策略

需要适当小心。

3.  ThreadPoolExecutor无界队列使用

public class ThreadPool {

private final static String poolName = "mypool";

static private ThreadPool threadFixedPool = new ThreadPool(2);

private ExecutorService executor;

static public ThreadPool getFixedInstance() {

return threadFixedPool;

}

private ThreadPool(int num) {

executor = Executors.newFixedThreadPool(num, new DaemonThreadFactory(poolName));

}

public void execute(Runnable r) {

executor.execute(r);

}

public static void main(String[] params) {

class MyRunnable implements Runnable {

public void run() {

System.out.println("OK!");

try {

Thread.sleep(10);

} catch (InterruptedException e) {

e.printStackTrace();

}

}

}

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

ThreadPool.getFixedInstance().execute(new MyRunnable());

}

try {

Thread.sleep(2000);

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

} catch (InterruptedException e) {

e.printStackTrace();

}

}

}

在这段代码中,我们发现我们用到了Executors.newFixedThreadPool()函数,这个函数的实现是这样子的:

return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>());

它实际上是创建了一个无界队列的固定大小的线程池。执行这段代码,我们发现所有的任务都正常处理了。但是在真实的线上环

境中会存在这样的一个问题,前端的用户请求源源不断的过来,后端的处理线程如果处理时间变长,无法快速的将用户请求处理

完返回结果给前端,那么任务队列中将堵塞大量的请求。这些请求在前端都是有超时时间设置的,假设请求是通过套接字过来,

当我们的后端处理进程处理完一个请求后,从队列中拿下一个任务,发现这个任务的套接字已经无效了,这是因为在用户端已经

超时,将套接字建立的连接关闭了。这样一来我们这边的处理程序再去读取套接字时,就会发生I/0 Exception. 恶性循环,导致我

们所有的处理服务线程读的都是超时的套接字,所有的请求过来都抛I/O异常,这样等于我们整个系统都挂掉了,已经无法对外提供

正常的服务了。

对于海量数据的处理,现在业界都是采用集群系统来进行处理,当请求的数量不断加大的时候,我们可以通过增加处理节点,反正现

在硬件设备相对便宜。但是要保证系统的可靠性和稳定性,在程序方面我们还是可以进一步的优化的,我们下一节要讲述的就是针对

线上出现的这类问题的一种处理策略。

4.   ThreadPoolExecutor有界队列使用

public class ThreadPool {

private final static String poolName = "mypool";

static private ThreadPool threadFixedPool = null;

public ArrayBlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(2);

private ExecutorService executor;

static public ThreadPool getFixedInstance() {

return threadFixedPool;

}

private ThreadPool(int num) {

executor = new ThreadPoolExecutor(2, 4,60,TimeUnit.SECONDS, queue,new DaemonThreadFactory

(poolName), new ThreadPoolExecutor.AbortPolicy());

}

public void execute(Runnable r) {

executor.execute(r);

}

public static void main(String[] params) {

class MyRunnable implements Runnable {

public void run() {

System.out.println("OK!");

try {

Thread.sleep(10);

} catch (InterruptedException e) {

e.printStackTrace();

}

}

}

int count = 0;

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

try {

ThreadPool.getFixedInstance().execute(new MyRunnable());

} catch (RejectedExecutionException e) {

e.printStackTrace();

count++;

}

}

try {

log.info("queue size:" + ThreadPool.getFixedInstance().queue.size());

Thread.sleep(2000);

} catch (InterruptedException e) {

e.printStackTrace();

}

System.out.println("Reject task: " + count);

}

}

首先我们来看下这段代码几个重要的参数,corePoolSize 为2,maximumPoolSize为4,任务队列大小为2,每个任务平

均处理时间为10ms,一共有10个并发任务。

执行这段代码,我们会发现,有4个任务失败了。这里就验证了我们在上面提到有界队列时候线程池的执行顺序。当新任务在

方法 execute(Runnable) 中提交时, 如果运行的线程少于 corePoolSize,则创建新线程来处理请求。 如果运行的线程多于

corePoolSize 而少于 maximumPoolSize,则仅当队列满时才创建新线程,如果此时线程数量达到maximumPoolSize,并且队

列已经满,就会拒绝继续进来的请求。

现在我们调整一下代码中的几个参数,将并发任务数改为200,执行结果Reject task: 182,说明有18个任务成功了,线程处理

完一个请求后会接着去处理下一个过来的请求。在真实的线上环境中,会源源不断的有新的请求过来,当前的被拒绝了,但只要线

程池线程把当下的任务处理完之后还是可以处理下一个发送过来的请求。

通过有界队列可以实现系统的过载保护,在高压的情况下,我们的系统处理能力不会变为0,还能正常对外进行服务,虽然有些服

务可能会被拒绝,至于如何减少被拒绝的数量以及对拒绝的请求采取何种处理策略我将会在下一篇文章《系统的过载保护》中继续

阐述。

时间: 2024-11-09 17:10:07

大数据处理系列之(一)Java线程池使用的相关文章

【java线程系列】java线程系列之java线程池详解

一线程池的概念及为何需要线程池: 我们知道当我们自己创建一个线程时如果该线程执行完任务后就进入死亡状态,这样如果我们需要在次使用一个线程时得重新创建一个线程,但是线程的创建是要付出一定的代价的,如果在我们的程序中需要频繁使用线程,且每个线程执行的时间很短,短到几乎小于线程创建及销毁的时间那么代价将会更大,如:服务器应用程序中经常出现的情况是:单个任务处理的时间很短而请求的数目却是巨大的.显然如果频繁的创建销毁线程效率将非常低. 那么我们能否让一个线程可以复用,即当一个线程执行完后不销毁该线程,而

大数据处理系列之(二)系统过载保护

前言:前段时间在网上看到腾讯后台开发总监bison分享的一篇文章<浅谈过载保护>,读来受益匪浅. 刚好自己也在处理系统请求过载的问题,把自己的一些心得体会总结出来拿来与大家一起探讨. 在bison的文章中谈到:对于延时敏感的服务,当外部请求超过系统处理能力,如果系统没有做相 应保护,可能导致历史累计的超时请求达到一定的规模,像雪球一样形成恶性循环,由于系统处理的每个 请求都因为超时而无效,系统对外呈现的服务能力为0,且这种情况不能自动恢复.我们的系统就是要尽 量避免这种情况的出现,下面将详细来

java线程池

1.new Thread的弊端 执行一个异步任务你还只是如下new Thread吗? 1 new Thread(new Runnable() { 2 3 @Override 4 public void run() { 5 // TODO Auto-generated method stub 6 } 7 }).start(); 那你就out太多了,new Thread的弊端如下: a. 每次new Thread新建对象性能差. b. 线程缺乏统一管理,可能无限制新建线程,相互之间竞争,及可能占用过

Java线程池使用说明

Java线程池使用说明 一 简介 线程的使用在java中占有极其重要的地位,在jdk1.4极其之前的jdk版本中,关于线程池的使用是极其简陋的.在jdk1.5之后这一情况有了很大的改观.Jdk1.5之后加入了java.util.concurrent包,这个包中主要介绍java中线程以及线程池的使用.为我们在开发中处理线程的问题提供了非常大的帮助. 二:线程池 线程池的作用: 线程池作用就是限制系统中执行线程的数量.     根 据系统的环境情况,可以自动或手动设置线程数量,达到运行的最佳效果:少

Java 线程池的原理与实现

最近在学习线程池.内存控制等关于提高程序运行性能方面的编程技术,在网上看到有一哥们写得不错,故和大家一起分享. [分享]Java 线程池的原理与实现 这几天主要是狂看源程序,在弥补了一些以前知识空白的同时,也学会了不少新的知识(比如 NIO),或者称为新技术吧.线程池就是其中之一,一提到线程,我们会想到以前<操作系统>的生产者与消费者,信号量,同步控制等等.一提到池,我们会想到数据库连接池,但是线程池又如何呢? 建议:在阅读本文前,先理一理同步的知识,特别是syncronized同步关键字的用

JAVA线程池的分析和使用

http://www.infoq.com/cn/articles/java-threadPool/ 1. 引言 合理利用线程池能够带来三个好处.第一:降低资源消耗.通过重复利用已创建的线程降低线程创建和销毁造成的消耗.第二:提高响应速度.当任务到达时,任务可以不需要等到线程创建就能立即执行.第三:提高线程的可管理性.线程是稀缺资源,如果无限制的创建,不仅会消耗系统资源,还会降低系统的稳定性,使用线程池可以进行统一的分配,调优和监控.但是要做到合理的利用线程池,必须对其原理了如指掌. 2. 线程池

java线程池分析和应用

比较 在前面的一些文章里,我们已经讨论了手工创建和管理线程.在实际应用中我们有的时候也会经常听到线程池这个概念.在这里,我们可以先针对手工创建管理线程和通过线程池来管理做一个比较.通常,我们如果手工创建线程,需要定义线程执行对象,它实现的接口.然后再创建一个线程对象,将我们定义好的对象执行部分装载到线程中.对于线程的创建.结束和结果的获取都需要我们来考虑.如果我们需要用到很多的线程时,对线程的管理就会变得比较困难.我们手工定义线程的方式在时间和空间效率方面会存在着一些不足.比如说我们定义好的线程

JAVA线程池ThreadPoolExecutor与阻塞队列BlockingQueue .

从Java5开始,Java提供了自己的线程池.每次只执行指定数量的线程,java.util.concurrent.ThreadPoolExecutor 就是这样的线程池.以下是我的学习过程. 首先是构造函数签名如下: [java] view plain copy print ? public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<

浅聊JAVA 线程池的一般用法

一.为什么要用线程池 1).降低资源消耗,通过重复利用已创建的线程降低线程的创建和销毁造成的消耗. 2).提高响应速度,当任务到达时,任务可以不需要等到线程创建就能立即执行. 3).提高线程的可管理性,线程是稀缺资源,如果无限制的创建,不仅会消耗系统资源,还会降低系统的稳定性,使用线程池可以进行统一的分配,调优和监控. 二.注意 1).需要对线程池原理了如指掌. 三.线程的常见用法 1).New Thread. 2).Thread Pool. 四.New Thread 常见用法 new  thr

Java线程池的几种实现 及 常见问题讲解

工作中,经常会涉及到线程.比如有些任务,经常会交与线程去异步执行.抑或服务端程序为每个请求单独建立一个线程处理任务.线程之外的,比如我们用的数据库连接.这些创建销毁或者打开关闭的操作,非常影响系统性能.所以,“池”的用处就凸显出来了. 1. 为什么要使用线程池 在3.6.1节介绍的实现方式中,对每个客户都分配一个新的工作线程.当工作线程与客户通信结束,这个线程就被销毁.这种实现方式有以下不足之处: 服务器创建和销毁工作的开销( 包括所花费的时间和系统资源 )很大.这一项不用解释,可以去查下"线程