java并发：线程池、饱和策略、定制、扩展

时间 2019-11-08

标签 java 并发线程饱和策略定制扩展栏目 Java 繁體版

原文原文链接

1、序言html

　　当咱们须要使用线程的时候，咱们能够新建一个线程，而后显式调用线程的start()方法，这样实现起来很是简便，但在某些场景下存在缺陷：若是须要同时执行多个任务（即并发的线程数量不少），频繁地建立线程会下降系统的效率，由于建立和销毁线程均须要必定的时间。java

　　线程池可使线程获得复用，所谓线程复用就是线程在执行完一个任务后并不被销毁，该线程能够继续执行其余的任务。java.lang.concurrent包中的Executors类为咱们建立线程池提供了方便。网络

2、Executors的简单使用示例并发

此处咱们先来看一个简单的例子，以下：ide

package com.soft;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ExecutorsDemo {
    
    public static void main(String[] args) throws InterruptedException, ExecutionException {
//      ExecutorService executor = Executors.newSingleThreadExecutor();
//      ExecutorService executor = Executors.newCachedThreadPool();
        ExecutorService executor = Executors.newFixedThreadPool(5);
        Thread.sleep(5*1000);//方便监控工具能捕获到
        for (int i = 0; i < 10; i++) {
            final int no = i;
            Runnable runnable = new Runnable() {
                public void run() {
                    try {
                        System.out.println("into" + no);
                        Thread.sleep(1000L);
                        System.out.println("end" + no);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            };
            executor.execute(runnable);//ExecutorService有一个execute()方法，这个方法的参数是Runnable类型，经过execute(Runnable)方法便可将一个任务添加到线程池，任务的执行方法是Runnable类型对象的run()方法。
        }//End for
        
        executor.shutdown();
        System.out.println("Thread Main End!");
    }
}

其运行结果以下：函数

into0
into3
Thread Main End!
into4
into1
into2
end0
into5
end3
end1
end4
into8
into6
into7
end2
into9
end5
end7
end8
end6
end9

解说：这个例子应该很容易看懂，从运行结果来看，在任意某一时刻只有5个线程在执行，这是由于上述代码经过Executors.newFixedThreadPool(5)语句建立了一个固定长度的线程池（长度为5），一个结束以后另再一个才开始执行。工具

3、Executors提供的线程池源码分析

　　Executors是线程的工厂类，也能够说是一个线程池工具类，它调用其内部静态方法（如newFixedThreadPool()等）便可建立一个线程池，经过参数设置，Executors提供不一样的线程池机制。测试

4、简述线程池的属性ui

5、详解ThreadPoolExecutor

　　上文提到能够经过显式的ThreadPoolExecutor构造函数来构造特定形式的线程池，ThreadPoolExecutor是java.util.concurrent包之内部线程池的形式对外提供线程池管理、线程调度等服务，此处咱们来了解一下ThreadPoolExecutor

（1）通常使用方式：

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

下文详解此示例涉及的一些内容

（2）构造函数的声明：

public ThreadPoolExecutor(int corePoolSize,  
                              int maximumPoolSize,  
                              long keepAliveTime,  
                              TimeUnit unit,  
                              BlockingQueue<Runnable> workQueue,  
                              ThreadFactory threadFactory,  
                              RejectedExecutionHandler handler) {  
        if (corePoolSize < 0 ||  
            maximumPoolSize <= 0 ||  
            maximumPoolSize < corePoolSize ||  
            keepAliveTime < 0)  
            throw new IllegalArgumentException();  
        if (workQueue == null || threadFactory == null || handler == null)  
            throw new NullPointerException();  
        this.corePoolSize = corePoolSize;  
        this.maximumPoolSize = maximumPoolSize;  
        this.workQueue = workQueue;  
        this.keepAliveTime = unit.toNanos(keepAliveTime);  
        this.threadFactory = threadFactory;  
        this.handler = handler;  
    }

（3）函数参数说明：

参数名	表明含义
corePoolSize	线程池的基本大小（核心线程池大小）
maximumPoolSize	线程池的最大大小
keepAliveTime	线程池中超过corePoolSize数目的空闲线程的最大存活时间
unit	keepAliveTime参数的时间单位
workQueue	任务阻塞队列
threadFactory	新建线程的工厂
handler	当提交的任务数超过maxmumPoolSize与workQueue之和时，任务会交给RejectedExecutionHandler来处理

进一步解说：

A、当提交新任务时，若线程池大小小于corePoolSize，将建立一个新的线程来执行任务，即便此时线程池中存在空闲线程；

B、当提交新任务时，若线程池达到corePoolSize大小，新提交的任务将被放入workQueue中，等待线程池调度执行；

C、当提交新任务时，若workQueue已满，且maximumPoolSize>corePoolSize，将建立新的线程来执行任务；

D、当提交新任务时，若任务总数超过maximumPoolSize，新提交的任务将由RejectedExecutionHandler来处理；

E、当线程池中的线程数超过corePoolSize时，若线程的空闲时间达到keepAliveTime，则关闭空闲线程

（4）任务阻塞队列选择机制

（5）简述SynchronousQueue

注：此处贴出SynchronousQueue的使用示例，示例中使用了Semaphore，更多关于SynchronousQueue及Semaphore的内容请参考其余文章

package com.test;

import java.util.concurrent.Semaphore;
import java.util.concurrent.SynchronousQueue;

/*
 * 程序中有10个线程来消费生成者产生的数据，这些消费者都调用TestDo.doSome()方法去进行处理，
 * 每一个消费者都须要一秒才能处理完，程序应保证这些消费者线程依次有序地消费数据，只有上一个消费者消费完后，
 * 下一个消费者才能消费数据，下一个消费者是谁均可以，但要保证这些消费者线程拿到的数据是有顺序的。
 */
public class SynchronousQueueTest {
    
    public static void main(String[] args) {
        
        System.out.println("begin:" + (System.currentTimeMillis() / 1000));
        // 定义一个Synchronous
        final SynchronousQueue<String> sq = new SynchronousQueue<String>(); // 定义一个数量为1的信号量，其做用至关于一个互斥锁
        final Semaphore sem = new Semaphore(1); for (int i = 0; i < 10; i++) {
            new Thread(new Runnable() {
                public void run() {
                    try {
                        sem.acquire();
                        String input = sq.take();
                        String output = TestDo.doSome(input);//内部类
                        System.out.println(Thread.currentThread().getName()+ ":" + output);
                        sem.release();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }).start();
        }
        
        for (int i = 0; i < 10; i++) {
            String input = i + ""; //此处将i变成字符串
            try {
                sq.put(input);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        
    }//End main
    
}

class TestDo {
    public static String doSome(String input) {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        String output = input + ":" + (System.currentTimeMillis() / 1000);
        return output;
    }
}

上述代码的运行结果以下：

begin:1458954798
Thread-0:0:1458954799
Thread-1:1:1458954800
Thread-2:2:1458954801
Thread-3:3:1458954802
Thread-4:4:1458954803
Thread-5:5:1458954804
Thread-6:6:1458954805
Thread-7:7:1458954806
Thread-8:8:1458954807
Thread-9:9:1458954808

从上述结果看，上例在任意某一时刻只有一个线程在执行，且只有前一个线程执行完下一个线程才开始

6、饱和策略（线程池任务拒绝策略）

上文提到ThreadPoolExecutor构造函数的RejectedExecutionHandler handler参数，该参数表示当提交的任务数超过maxmumPoolSize与workQueue之和时，任务会交给RejectedExecutionHandler来处理，此处咱们来具体了解一下

（1）四种饱和策略

（2）源码分析：

RejectedExecutionHandler这个接口是用来处理被丢弃的线程的异常处理接口，其源码以下：

public interface RejectedExecutionHandler{
    //被线程池丢弃的线程处理机制
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) ;
}

AbortPolicy（停止策略）继承RejectedExecutionHandler接口，其源码以下：

public static class AbortPolicy implements RejectedExecutionHandler{

    public AbortPolicy(){}
    
    //直接抛出异常
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        throw new RejectedExecutionException("Task"+r.toString()+"rejected from"+executor.toString());
    }
    
}

咱们能够本身实现RejectedExecutionHandler接口，将实现类做为线程丢弃处理类，代码以下：

package com.test;

import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;

public class RejectedExecutionHandlerDemo implements RejectedExecutionHandler{

    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        // TODO Auto-generated method stub
        System.out.println("线程信息"+r.toString()+"被遗弃的线程池："+executor.toString());
    }
    
}

7、定制ThreadPoolExecutor

（1）经过修改参数的方式达到定制目的

（2）经过自定义方式（封装各类参数）达到定制目的

示例（摘自网络）：

import java.util.concurrent.ArrayBlockingQueue;  
import java.util.concurrent.ExecutorService;  
import java.util.concurrent.RejectedExecutionHandler;  
import java.util.concurrent.ThreadFactory;  
import java.util.concurrent.ThreadPoolExecutor;  
import java.util.concurrent.TimeUnit;  
import java.util.concurrent.atomic.AtomicInteger;  
  
  
public class CustomThreadPoolExecutor {  
  
      
    private ThreadPoolExecutor pool = null;  
      
      
    /** 
     * 线程池初始化方法 
     *  
     * corePoolSize 核心线程池大小----10 
     * maximumPoolSize 最大线程池大小----30 
     * keepAliveTime 线程池中超过corePoolSize数目的空闲线程最大存活时间----30+单位TimeUnit 
     * TimeUnit keepAliveTime时间单位----TimeUnit.MINUTES 
     * workQueue 阻塞队列----new ArrayBlockingQueue<Runnable>(10)====10容量的阻塞队列 
     * threadFactory 新建线程工厂----new CustomThreadFactory()====定制的线程工厂 
     * rejectedExecutionHandler 当提交任务数超过maxmumPoolSize+workQueue之和时, 
     *                          即当提交第41个任务时(前面线程都没有执行完,此测试方法中用sleep(100)), 
     *                          任务会交给RejectedExecutionHandler来处理 
     */  
    public void init() {  
        pool = new ThreadPoolExecutor(  
                10,  
                30,  
                30,  
                TimeUnit.MINUTES,  
                new ArrayBlockingQueue<Runnable>(10),  
                new CustomThreadFactory(),new CustomRejectedExecutionHandler());  
    }  
  
      
    public void destory() {  
        if(pool != null) {  
            pool.shutdownNow();  
        }  
    }  
      
      
    public ExecutorService getCustomThreadPoolExecutor() {  
        return this.pool;  
    }  
      
    private class CustomThreadFactory implements ThreadFactory {  
  
        private AtomicInteger count = new AtomicInteger(0);  
          
        @Override  
        public Thread newThread(Runnable r) {  
            Thread t = new Thread(r);  
            String threadName = CustomThreadPoolExecutor.class.getSimpleName() + count.addAndGet(1);  
            System.out.println(threadName);  
            t.setName(threadName);  
            return t;  
        }  
    }  
      
      
    private class CustomRejectedExecutionHandler implements RejectedExecutionHandler {  
  
        @Override  
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {  
            // 记录异常  
            // 报警处理等  
            System.out.println("error.............");  
        }  
    }  
      
      
      
    // 测试构造的线程池  
    public static void main(String[] args) {  
        CustomThreadPoolExecutor exec = new CustomThreadPoolExecutor();  
        // 1.初始化  
        exec.init();  
          
        ExecutorService pool = exec.getCustomThreadPoolExecutor();  
        for(int i=1; i<100; i++) {  
            System.out.println("提交第" + i + "个任务!");  
            pool.execute(new Runnable() {  
                @Override  
                public void run() {  
                    try {  
                        Thread.sleep(300);  
                    } catch (InterruptedException e) {  
                        e.printStackTrace();  
                    }  
                    System.out.println("running=====");  
                }  
            });  
        }  
          
          
          
        // 2.销毁----此处不能销毁,由于任务没有提交执行完,若是销毁线程池,任务也就没法执行了  
        // exec.destory();  
          
        try {  
            Thread.sleep(10000);  
        } catch (InterruptedException e) {  
            e.printStackTrace();  
        }  
    }  
}

咱们能够看到上述代码是经过init()方法对ThreadPoolExecutor构造函数进行了一些自定义设置，getCustomThreadPoolExecutor()方法返回init()方法配置的ThreadPoolExecutor对象实例（即线程池引用）

补充：

ThreadPoolExecutor构造函数有一个参数ThreadFactory threadFactory，前文提到该参数是新建线程的工厂，此处进一步解说该参数。

ThreadFactory是java.util.concurrent包下建立线程工厂的接口，ThreadFactory接口源码以下：

public interface ThreadFactory {  
    Thread newThread(Runnable r);  
}

　　JDK线程池：Executors.newSingleThreadExecutor、Executors.newFixedThreadPool等由一个ThreadFactory来建立新的线程，默认状况下为Executors.defaultThreadFactory()，咱们能够采用自定义的ThreadFactory工厂，增长对线程建立与销毁等更多的控制（好比上述代码中的内部类CustomThreadFactory即为新建线程的模板）

此处简单说起一下，读者欲了解更多内容能够参考如下文章

（1）http://guojuanjun.blog.51cto.com/277646/650981/

（2）http://ifeve.com/customizing-concurrency-classes-4/

8、扩展ThreadPoolExecutor

9、源码视角

从源码视角分析Executors、ThreadPoolExecutor、ExecuteService、Executor之间的关系，此处简单说起，读者可查看下一节“参考资料”以了解相关内容

（1）Executors

　　从Java5开始新增了Executors类，它有几个静态工厂方法用来建立线程池，这些静态工厂方法返回一个ExecutorService类型的值，此值即为线程池的引用。

（2）Executor

Executor是一个接口，里面只有一个方法

public interface Executor {
    void execute(Runnable command);
}

（3）ExecuteService

ExecuteService也是一个接口，其定义以下：

public interface ExecutorService extends Executor {...}

（4）ThreadPoolExecutor继承AbstractExecutorService，AbstractExecutorService实现ExecutorService接口

public class ThreadPoolExecutor extends AbstractExecutorService {...}

public abstract class AbstractExecutorService implements ExecutorService {...}

10、ExecutorService的生命周期

在本文最开始的那个示例中，有一句代码，以下：

executor.shutdown();

该语句并非终止线程的运行，而是禁止在这个executor中添加新的任务，下文描述了该语句对于ExecutorService的意义。

11、参考资料

本文仅简单阐述了Java并发中关于Executors及ThreadPoolExecutor的内容，此处贴出一些优质文章以供读者阅览

（1）http://blog.csdn.net/xiamizy/article/details/40781939

（2）http://www.cnblogs.com/dolphin0520/p/3932921.html

（3）http://www.cnblogs.com/yezhenhan/archive/2012/01/07/2315645.html

（4）http://www.cnblogs.com/guguli/p/5198894.html