Concurrent - 线程池

原创转载请注明出处：http://agilestyle.iteye.com/blog/2390819

 

ThreadPoolExecutor底层方法参数：

@param corePoolSize: the number of threads to keep in the pool, even if they are idle, unless {@code allowCoreThreadTimeOut} is set

@param maximumPoolSize: the maximum number of threads to allow in the pool

@param keepAliveTime: when the number of threads is greater than the core, this is the maximum time that excess idle threads will wait for new tasks before terminating.

@param unit: the time unit for the {@code keepAliveTime} argument

@param workQueue: the queue to use for holding tasks before they are executed.  This queue will hold only the {@code Runnable} tasks submitted by the {@code execute} method.

@param threadFactory: the factory to use when the executor creates a new thread

@param handler: the handler to use when execution is blocked because the thread bounds and queue capacities are reached 

 

Reject策略预定义有四种： 

(1)ThreadPoolExecutor.AbortPolicy策略，是默认的策略,处理程序遭到拒绝将抛出运行时 RejectedExecutionException。 

(2)ThreadPoolExecutor.CallerRunsPolicy策略 ,调用者的线程会执行该任务,如果执行器已关闭,则丢弃. 

(3)ThreadPoolExecutor.DiscardPolicy策略，不能执行的任务将被丢弃. 

(4)ThreadPoolExecutor.DiscardOldestPolicy策略，如果执行程序尚未关闭，则位于工作队列头部的任务将被删除，然后重试执行程序（如果再次失败，则重复此过程）.

 

ThreadPoolExecutor执行器的处理流程: 

(1)当线程池大小小于corePoolSize就新建线程，并处理请求. 

(2)当线程池大小等于corePoolSize，把请求放入workQueue中，池子里的空闲线程就去从workQueue中取任务并处理. 

(3)当workQueue放不下新入的任务时，新建线程加入线程池，并处理请求，如果池子大小撑到了maximumPoolSize就用RejectedExecutionHandler来做拒绝处理. 

(4)另外，当线程池的线程数大于corePoolSize的时候，多余的线程会等待keepAliveTime长的时间，如果无请求可处理就自行销毁.

package org.fool.test.thread;

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class ThreadPoolExecutorTest {
    public static void main(String[] args) {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 3, 0, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(5), new ThreadFactory() {
            private final AtomicInteger threadNum = new AtomicInteger(0);

            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "thread-" + threadNum.getAndIncrement());
            }
        });

        for (int i = 0; i < 8; i++) {
            int index = i;

            executor.execute(() -> {
                try {
                    Thread.sleep(3000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println(Thread.currentThread().getName() + " invoked task:" + index);
            });
        }
    }
} 

例子中 corePoolSize:1, maximumPoolSize: 3, 有界队列容量是5

查看结果输出，可以看到总共执行8个任务0到7，超过了有界队列的容量5，线程1和线程2就优先处理超过有界队列容量的任务6和任务7

 

Spring Bean Example:

<bean id="threadPoolTaskExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
	<!-- 线程池维护线程的最少数量 -->
	<property name="corePoolSize" value="30"/>
	<!-- 线程池维护线程的最大数量 -->
	<property name="maxPoolSize" value="400"/>
	<!-- 缓存队列 -->
	<property name="queueCapacity" value="100"/>
	<!-- 允许的空闲时间 -->
	<property name="keepAliveSeconds" value="200"/>
	<!-- 对拒绝task的处理策略 -->
	<property name="rejectedExecutionHandler">
		<bean class="java.util.concurrent.ThreadPoolExecutor$CallerRunsPolicy"/>
	</property>
</bean>