欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

学习JAVA线程池

程序员文章站 2024-03-02 14:58:40
...

一、线程池作用
线程池作用就是限制系统中执行线程的数量。
根据系统的环境情况,可以自动或手动设置线程数量,达到运行的最佳效果;少了浪费了系统资源,多了造成系统拥挤效率不高。用线程池控制线程数量,其他线程排队等候。一个任务执行完毕,再从队列的中取最前面的任务开始执行。若队列中没有等待进程,线程池的这一资源处于等待。当一个新任务需要运行时,如果线程池中有等待的工作线程,就可以开始运行了;否则进入等待队列。

基于生产者-消费者模式,其提交任务的线程相当于生产者,执行任务的线程相当于消费者,并用Runnable来表示任务

二、重要接口和类
学习JAVA线程池

1.接口Executor
学习JAVA线程池
定义了一个接收Runnable对象的方法execute

2.接口ExecutorService
学习JAVA线程池
是一个比Executor使用更广泛的子类接口,其提供了生命周期(运行、关闭、已终止)管理的方法,以及可跟踪一个或多个异步任务执行状况返回Future的方法。Future 表示一个任务的生命周期,并且提供了相应的方法来判断是否已经完成或取消,以及获取任务的结果和取消任务等。
创建一个什么样的ExecutorService的实例(即线程池)需要根据具体应用场景而定,不过Java给我们提供了一个Executors工厂类,它可以帮助我们很方便的创建各种类型ExecutorService线程池。

2.1 ExecutorService的使用

ExecutorService executorService = Executors.newFixedThreadPool(10);

executorService.execute(new Runnable() {
    public void run() {
        System.out.println("Asynchronous task");
    }
});

executorService.shutdown();

2.2 ExecutorService的执行
ExecutorService有如下几个执行方法:

- execute(Runnable)
- submit(Runnable)
- submit(Callable)
- invokeAny(...)
- invokeAll(...)

2.2.1 execute(Runnable)
这个方法接收一个Runnable实例,并且异步的执行,请看下面的实例:

ExecutorService executorService = Executors.newSingleThreadExecutor();

executorService.execute(new Runnable() {
    public void run() {
        System.out.println("Asynchronous task");
    }
});

executorService.shutdown();

这个方法有个问题,就是没有办法获知task的执行结果。如果我们想获得task的执行结果,我们可以传入一个Callable的实例(下面会介绍)。

2.2.2 submit(Runnable)
submit(Runnable)和execute(Runnable)区别是前者可以返回一个Future对象,通过返回的Future对象,我们可以检查提交的任务是否执行完毕,请看下面执行的例子:

Future future = executorService.submit(new Runnable() {
    public void run() {
        System.out.println("Asynchronous task");
    }
});

future.get();  //returns null if the task has finished correctly.

如果任务执行完成,future.get()方法会返回一个null。注意,future.get()方法会产生阻塞。

2.2.3 submit(Callable)
submit(Callable)和submit(Runnable)类似,也会返回一个Future对象,但是除此之外,submit(Callable)接收的是一个Callable的实现,Callable接口中的call()方法有一个返回值,可以返回任务的执行结果,而Runnable接口中的run()方法是void的,没有返回值。请看下面实例:

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());

如果任务执行完成,future.get()方法会返回Callable任务的执行结果。注意,future.get()方法会产生阻塞,直到call方法结束返回结果。

2.2.4 invokeAny(…)
invokeAny(…)方法接收的是一个Callable的集合,执行这个方法不会返回Future,但是会返回所有Callable任务中其中一个任务的执行结果。这个方法也无法保证返回的是哪个任务的执行结果,反正是其中的某一个。请看下面实例:

ExecutorService executorService = Executors.newSingleThreadExecutor();

Set<Callable<String>> callables = new HashSet<Callable<String>>();

callables.add(new Callable<String>() {
    public String call() throws Exception {
        return "Task 1";
    }
});
callables.add(new Callable<String>() {
    public String call() throws Exception {
        return "Task 2";
    }
});
callables.add(new Callable<String>() {
    public String call() throws Exception {
        return "Task 3";
    }
});

String result = executorService.invokeAny(callables);
System.out.println("result = " + result);
executorService.shutdown();

大家可以尝试执行上面代码,每次执行都会返回一个结果,并且返回的结果是变化的,可能会返回“Task2”也可是“Task1”或者其它。

2.2.5 invokeAll(…)
invokeAll(…)与 invokeAny(…)类似也是接收一个Callable集合,但是前者执行之后会返回一个Future的List,其中对应着每个Callable任务执行后的Future对象。情况下面这个实例:

ExecutorService executorService = Executors.newSingleThreadExecutor();

Set<Callable<String>> callables = new HashSet<Callable<String>>();

callables.add(new Callable<String>() {
    public String call() throws Exception {
        return "Task 1";
    }
});
callables.add(new Callable<String>() {
    public String call() throws Exception {
        return "Task 2";
    }
});
callables.add(new Callable<String>() {
    public String call() throws Exception {
        return "Task 3";
    }
});

List<Future<String>> futures = executorService.invokeAll(callables);

for(Future<String> future : futures){
    System.out.println("future.get = " + future.get());
}

executorService.shutdown();

2.3 ExecutorService的关闭
当我们使用完成ExecutorService之后应该关闭它,否则它里面的线程会一直处于运行状态。

如果要关闭ExecutorService中执行的线程,我们可以调用ExecutorService.shutdown()方法。在调用shutdown()方法之后,ExecutorService不会立即关闭,但是它不再接收新的任务,直到当前所有线程执行完成才会关闭,所有在shutdown()执行之前提交的任务都会被执行。

如果我们想立即关闭ExecutorService,我们可以调用ExecutorService.shutdownNow()方法。这个动作将跳过所有正在执行的任务和被提交还没有执行的任务。但是它并不对正在执行的任务做任何保证,有可能它们都会停止,也有可能执行完成。

3.接口ScheduledExecutorService
学习JAVA线程池

3.1 scheduleAtFixedRate() 按指定频率周期执行某个任务

public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, 
            long initialDelay,  
            long period,  
            TimeUnit unit); 

command:执行线程
initialDelay:初始化延时
period:两次开始执行最小间隔时间
unit:计时单位
实例:

private static ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor();  
static {
       service.scheduleAtFixedRate(new Runnable() {  
           public void run() {  
               System.out.println("延迟10s开始执行,每隔100s重新执行一次任务");
           }  
       }, 10, 100, TimeUnit.SECONDS); 
}

3.2 scheduleWithFixedDelay() 按指定频率间隔执行某个任务

public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,  
                long initialDelay,  
                long delay,  
                TimeUnit unit); 

command:执行线程
initialDelay:初始化延时
period:前一次执行结束到下一次执行开始的间隔时间(间隔执行延迟时间)
unit:计时单位
实例:

    private static ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor();  
    static {
        service.scheduleWithFixedDelay(new Runnable() {  
            public void run() {  
                System.out.println("10秒后开始执行,本次执行结束后延迟100s开始下次执行。");
            }  
        }, 10, 100, TimeUnit.SECONDS); 
    }

3.3 周期定时执行某个任务

/** 
 * 每天晚上8点执行一次 
 * 每天定时安排任务进行执行 
 */  
public static void executeEightAtNightPerDay() {  
    ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);  
    long oneDay = 24 * 60 * 60 * 1000;  
    long initDelay  = getTimeMillis("20:00:00") - System.currentTimeMillis();  
    initDelay = initDelay > 0 ? initDelay : oneDay + initDelay;  

    executor.scheduleAtFixedRate(new Runnable() {  
        public void run() {  
            System.out.println("do something");
        },  
        initDelay,  
        oneDay,  
        TimeUnit.MILLISECONDS);  
}  
/** 
 * 获取指定时间对应的毫秒数 
 * @param time "HH:mm:ss" 
 * @return 
 */  
private static long getTimeMillis(String time) {  
    try {  
        DateFormat dateFormat = new SimpleDateFormat("yy-MM-dd HH:mm:ss");  
        DateFormat dayFormat = new SimpleDateFormat("yy-MM-dd");  
        Date curDate = dateFormat.parse(dayFormat.format(new Date()) + " " + time);  
        return curDate.getTime();  
    } catch (ParseException e) {  
        e.printStackTrace();  
    }  
    return 0;  
}  

4.类Executors
通过 Executors 中的静态工厂方法来创建线程池。

4.1 newFixedThreadPool()
public static ExecutorService newFixedThreadPool(int nThreads)

创建一个定长线程池,定长线程池的大小最好根据系统资源进行设置,可控制线程最大并发数,超出的线程会在队列中等待。
每次提交一个任务就创建一个线程,直到线程达到线程池的最大大小。线程池的大小一旦达到最大值就会保持不变,如果某个线程因为执行异常而结束,那么线程池会补充一个新线程。
newFixedThreadPool内部有个任务队列,假设线程池里有3个线程,提交了5个任务,那么后两个任务就放在任务队列了,即使前3个任务sleep或者堵塞了,也不会执行后两个任务,除非前三个任务有执行完的

实例:

public class Test {
    public static void main(String[] args) throws IOException, InterruptedException {
        ExecutorService service = Executors.newFixedThreadPool(2);
        for (int i = 0; i < 6; i++) {
            final int index = i;
            System.out.println("task: " + (i + 1));
            Runnable run = new Runnable() {
                @Override
                public void run() {
                    System.out.println("thread start" + index);
                    try {
                        Thread.sleep(Long.MAX_VALUE);//执行很久。。。
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println("thread end" + index);
                }
            };
            service.execute(run);
        }
        service.shutdown();
    }
}

输出:

task: 1
task: 2
thread start0
thread start1
task: 3
task: 4
task: 5
task: 6

6个任务,只有二个任务在执行,4个在任务队列,等候执行。

4.2 newCachedThreadPool()
创建一个可缓存线程池,如果线程池中的线程在60秒未被使用就将被移除,在执行新的任务时,当线程池中有之前创建的可用线程就重用可用线程,否则就新建一条线程

public class Test {
    public static void main(String[] args) {
        ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
        for (int i = 0; i < 10; i++) {
            final int index = i;
            try {
                Thread.sleep(index * 1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            cachedThreadPool.execute(new Runnable() {
                @Override
                public void run() {
                    System.out.println(index);
                }
            });
        }
    }
}

线程池为无限大,当执行第二个任务时第一个任务已经完成,会复用执行第一个任务的线程,而不用每次新建线程。
对于执行很多短期异步任务的程序而言,可提高程序性能。

4.3 newScheduledThreadPool()
创建一个定长线程池,支持定时及周期性任务执行。实例同【3.接口ScheduledExecutorService】

4.4 newSingleThreadExecutor()
创建一个单线程化的线程池,它只会用唯一的工作线程来执行任务,保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。

public class Test {
    public static void main(String[] args) {
        ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
        for (int i = 0; i < 10; i++) {
            final int index = i;
            singleThreadExecutor.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        System.out.println(index);
                        Thread.sleep(2000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
        }
    }
}

结果依次输出,相当于顺序执行各个任务。

5.类ThreadPoolExecutor

private static ExecutorService exec = new ThreadPoolExecutor(8, 8, 0L,
        TimeUnit.MILLISECONDS, 
        new LinkedBlockingQueue<Runnable>(100000),
        new ThreadPoolExecutor.CallerRunsPolicy());

5.1 常用构造方法

ThreadPoolExecutor(int corePoolSize, 
        int maximumPoolSize, 
        long keepAliveTime, 
        TimeUnit unit, 
        BlockingQueue<Runnable> workQueue, 
        RejectedExecutionHandler handler)
corePoolSize 线程(核心)池维护线程的最少数量
maximumPoolSize 线程(最大)池维护线程的最大数量
keepAliveTime 线程池维护线程所允许的空闲时间
unit 线程池维护线程所允许的空闲时间的单位
workQueue 线程池所使用的缓冲队列
handler 线程池对拒绝任务的处理策略



5.1.1
一个任务通过 execute(Runnable)方法被添加到线程池,任务就是一个 Runnable类型的对象,任务的执行方法就是Runnable类型对象的run()方法。当一个任务通过execute(Runnable)方法欲添加到线程池时:

l 如果此时线程池中的数量小于corePoolSize,即使线程池中的线程都处于空闲状态,也要创建新的线程来处理被添加的任务。

l 如果此时线程池中的数量等于 corePoolSize,但是缓冲队列 workQueue未满,那么任务被放入缓冲队列。

l 如果此时线程池中的数量大于corePoolSize,缓冲队列workQueue满,并且线程池中的数量小于maximumPoolSize,建新的线程来处理被添加的任务。

l 如果此时线程池中的数量大于corePoolSize,缓冲队列workQueue满,并且线程池中的数量等于maximumPoolSize,那么通过 handler所指定的策略来处理此任务。也就是:处理任务的优先级为:核心线程corePoolSize、任务队列workQueue、最大线程maximumPoolSize,如果三者都满了,使用handler处理被拒绝的任务。

l 当线程池中的线程数量大于 corePoolSize时,如果某线程空闲时间超过keepAliveTime,线程将被终止。这样,线程池可以动态的调整池中的线程数。

5.1.2 handler有四个选择

默认new ThreadPoolExecutor.AbortPolicy() 处理程序遭到拒绝将抛出运行时RejectedExecutionException
new ThreadPoolExecutor.CallerRunsPolicy() 当抛出RejectedExecutionException异常时,会调用rejectedExecution方法
new ThreadPoolExecutor.DiscardOldestPolicy() 抛弃旧的任务,如果执行程序尚未关闭,则位于工作队列头部的任务将被删除,然后重试执行程序(如果再次失败,则重复此过程)
new ThreadPoolExecutor.DiscardPolicy() 抛弃当前的任务,不能执行的任务将被删除



实例:(实例原文http://blog.csdn.net/pfnie/article/details/52755769
写一个task:

public class Task implements Runnable {  

    protected String name;  

    public Task(String name) {  
        super();  
        this.name = name;  
    }  

    @Override  
    public void run() {  
        try {  
            System.out.println(this.name + " is running.");    
            Thread.sleep(500);   
        } catch (Exception e) {  

        }  
    }  
}  

1) AbortPolicy 示例

import java.util.concurrent.ArrayBlockingQueue;  
import java.util.concurrent.RejectedExecutionException;  
import java.util.concurrent.ThreadPoolExecutor;  
import java.util.concurrent.TimeUnit;  

public class AbortPolicyDemo {  

    public static void main(String[] args) {  
        // 创建线程池。线程池的"最大池大小"和"核心池大小"都为1,"线程池"的阻塞队列容量为1。    
        ThreadPoolExecutor pool = new ThreadPoolExecutor(1, 1, 0,  
                TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(1));  

        // 设置线程池的拒绝策略为AbortPolicy    
        pool.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy());
        // 设置线程池的拒绝策略为CallerRunsPolicy    
        // pool.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());  
        // 设置线程池的拒绝策略为DiscardPolicy    
        // pool.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardPolicy()); 
        // 设置线程池的拒绝策略为DiscardOldestPolicy    
        // pool.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardOldestPolicy());          
        try {  
            // 新建10个任务,并将它们添加到线程池中    
            for (int i = 0; i < 10; i++) {    
                Runnable myTask = new Task("task-"+i);    
                pool.execute(myTask);    
            }    
        } catch (RejectedExecutionException e) {  
            e.printStackTrace();  
             // 关闭线程池    
            pool.shutdown();  
        }  
    }  
} 

某一次运行结果,直接已异常的形式抛出:

task-0 is running.

java.util.concurrent.RejectedExecutionException: Task websocket.blocktest.Task@58777255 rejected from java.util.concurrent.ThreadPoolExecutor@470ae2bf[Running, pool size = 1, active threads = 1, queued tasks = 1, completed tasks = 0]
    at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2048)
    at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:821)
    at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1372)
    at websocket.blocktest.Test.main(Test.java:22)

task-1 is running.



2)CallerRunsPolicy 示例

// 设置线程池的拒绝策略为CallerRunsPolicy    
pool.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); 

某一次运行结果,当有新任务添加到线程池被拒绝时,线程池会将被拒绝的任务添加到”线程池正在运行的线程”中去运行:

task-2 is running.
task-0 is running.
task-3 is running.
task-1 is running.
task-5 is running.
task-4 is running.
task-7 is running.
task-6 is running.
task-9 is running.
task-8 is running.



3)DiscardPolicy 示例

// 设置线程池的拒绝策略为DiscardPolicy    
pool.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardPolicy()); 

某一次运行结果,线程池pool的”最大池大小”和”核心池大小”都为1,这意味着”线程池能同时运行的任务数量最大只能是1”。线程池pool的阻塞队列是ArrayBlockingQueue,ArrayBlockingQueue是一个有界的阻塞队列,ArrayBlockingQueue的容量为1。这也意味着线程池的阻塞队列只能有一个线程池阻塞等待。由此可知,线程池*运行了2个任务。第1个任务直接放到Worker中,通过线程去执行;第2个任务放到阻塞队列中等待。其他的任务都被丢弃了。

task-0 is running.
task-1 is running.



4)DiscardOldestPolicy 示例

// 设置线程池的拒绝策略为DiscardOldestPolicy    
pool.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardOldestPolicy());

某一次运行结果,当有新任务添加到线程池被拒绝时,线程池会丢弃阻塞队列中末尾的任务,然后将被拒绝的任务添加到末尾。

task-0 is running.
task-9 is running.


5.2 ThreadPoolExecutor配置(加深理解)

5.2.1 线程的创建与销毁
1)核心池大小、最大池大小和存活时间共同管理着线程的创建与销毁。
2)核心池的大小是目标的大小;线程池的实现试图维护池的大小;即使没有任务执行,池的大小也等于核心池的大小,并直到工作队列充满前,池都不会创建更多的线程。如果当前池的大小超过了核心池的大小,线程池就会终止它。
3)最大池的大小是可同时活动的线程数的上限。
4)如果一个线程已经闲置的时间超过了存活时间,它将成为一个被回收的候选者。
5)newFixedThreadPool工厂为请求的池设置了核心池的大小和最大池的大小,而且池永远不会超时
6)newCacheThreadPool工厂将最大池的大小设置为Integer.MAX_VALUE,核心池的大小设置为0,超时设置为一分钟。这样创建了无限扩大的线程池,会在需求量减少的情况下减少线程数量。

5.2.2 管理
1)ThreadPoolExecutor允许你提供一个BlockingQueue来持有等待执行的任务。任务排队有3种基本方法:无限队列、有限队列和同步移交。
2)newFixedThreadPool和newSingleThreadExectuor默认使用的是一个无限的 LinkedBlockingQueue。如果所有的工作者线程都处于忙碌状态,任务会在队列中等候。如果任务持续快速到达,超过了它们被执行的速度,队列也会无限制地增加。稳妥的策略是使用有限队列,比如ArrayBlockingQueue或有限的LinkedBlockingQueue以及 PriorityBlockingQueue。
3)对于庞大或无限的池,可以使用SynchronousQueue,完全绕开队列,直接将任务由生产者交给工作者线程
4)可以使用PriorityBlockingQueue通过优先级安排任务

6.类ScheduledThreadPoolExecutor
平时我们在执行一个定时任务时,会采用Time和TimeTask来组合处理,但是Timer和TimerTask存在一些缺陷:
1)Timer只创建了一个线程。当你的任务执行的时间超过设置的延时时间将会产生一些问题。
2)Timer创建的线程没有处理异常,因此一旦抛出非受检异常,该线程会立即终止。
JDK 5.0以后推荐使用ScheduledThreadPoolExecutor。该类属于Executor Framework,它除了能处理异常外,还可以创建多个线程解决上面的问题。

实例:

public class Test {
    public static void main(String[] args) throws InterruptedException {
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        ScheduledThreadPoolExecutor pool = new ScheduledThreadPoolExecutor(5);
        for (int i = 0; i < 5; i ++){
            final int temp = i + 1;
            //jdk8 lambda表达式
            pool.schedule(() -> {
                System.out.println("第"+temp+"个炸弹爆炸时间:" + simpleDateFormat.format(new Date()));
            }, temp * 5, TimeUnit.SECONDS);

          //隔2秒后开始执行任务,并且在上一次任务开始后隔一秒再执行一次;  
          //pool.scheduleWithFixedDelay(new MyTask(), 2, 1, TimeUnit.SECONDS);  
          //隔6秒后执行一次,但只会执行一次。  
          //pool.schedule(new MyTask(), 6, TimeUnit.SECONDS);  
        }
        pool.shutdown();
        System.out.println("end main时间:" + simpleDateFormat.format(new Date()));
    }

    private static class MyTask implements Runnable{
        @Override
        public void run() { 
            //捕获所有的异常,保证定时任务能够继续执行
            try{
                System.out.println("任务开始...");
                // 业务...
                System.out.println("任务结束...");
            }catch (Throwable e) {
                // 什么也不做
            }
        }
    }
}

在JAVAWEB开发中,执行定时任务有一个更好的选择: Quartz