線程池工具類的封裝

瞭解更多學習 ThreadPoolExecutor類

ThreadPool.java

package com.tool.me.thread;

import java.util.Hashtable;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class ThreadPool {
	private static Map<String, ThreadPoolExecutor> map = new Hashtable<>();
	private ThreadPoolExecutor executor;
	/** * 阻塞任務隊列數 */
	private int wattingCount;
	/** * 線程池的名字，e.g:子系統的包名(com.tool.me) */
	@SuppressWarnings("unused")
	private String name;

	/** * 建立線程池 * * @param name * 線程池的名字，eg:子系統的包名(com.tool.me) * @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 * 線程池中超過corePoolSize數目的空閒線程最大存活時間 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 * keepAliveTime時間單位 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. */
	public ThreadPool(String name, int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
		synchronized (map) {
			this.name = name;
			this.wattingCount = workQueue.size();
			String key = buildKey(name, corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue.size(), "#");
			if (map.containsKey(key)) {
				executor = map.get(key);
			} else {
				executor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
				map.put(key, executor);
			}
		}
	}

	/** * 建立線程池 * * @param name * 線程池的名字，eg:子系統的包名(com.tool.me) * @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 * 線程池中超過corePoolSize數目的空閒線程最大存活時間 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 * keepAliveTime時間單位 the time unit for the {@code keepAliveTime} * argument * @param wattingCount * 阻塞任務隊列數 */
	public ThreadPool(String name, int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, int wattingCount) {
		synchronized (map) {
			this.name = name;
			this.wattingCount = (int) (wattingCount * 1.5);
			String key = buildKey(name, corePoolSize, maximumPoolSize, keepAliveTime, unit, wattingCount, "#");
			if (map.containsKey(key)) {
				executor = map.get(key);
			} else {
				executor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit,
						new LinkedBlockingQueue<Runnable>(this.wattingCount));
				map.put(key, executor);
			}
		}
	}

	/** * 組裝map中的key * * @param name * 線程池的名字，eg:子系統的包名(com.tool.me) * @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 * 線程池中超過corePoolSize數目的空閒線程最大存活時間 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 * keepAliveTime時間單位 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 delimiter * 分割符 */
	private String buildKey(String name, int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, int wattingCount, String delimiter) {
		StringBuilder result = new StringBuilder();
		result.append(name).append(delimiter);
		result.append(corePoolSize).append(delimiter);
		result.append(maximumPoolSize).append(delimiter);
		result.append(keepAliveTime).append(delimiter);
		result.append(unit.toString()).append(delimiter);
		result.append(wattingCount);
		return result.toString();
	}
	
	/** * 添加任務到線程池（execute）中 * @param runnable the task to execute */
	public void execute(Runnable runnable) {
		checkQueneSize();
		executor.execute(runnable);
	}
	
	private void checkQueneSize() {
		while (getTaskSzie() >= wattingCount) {//若是線程池中的阻塞隊列數 > wattingCount 則繼續等待
			try {
				Thread.sleep(100);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}
	
	/** * Returns the number of elements in this collection. If this collection * contains more than <tt>Integer.MAX_VALUE</tt> elements, returns * <tt>Integer.MAX_VALUE</tt>. * * @return the number of elements in this collection */
	public int getTaskSzie(){
		return executor.getQueue().size();
	}
}
複製代碼

ThreadPoolManager.java

package com.tool.me.thread;

import java.util.concurrent.TimeUnit;

public abstract class ThreadPoolManager {
	private ThreadPool executor = null;
	
	public ThreadPoolManager() {
		if(executor == null) {
			executor = new ThreadPool(getThreadPoolName(),corePoolSize(), maximumPoolSize(), keepAliveTime(), TimeUnit.SECONDS, wattingCount());
		}
	}
	
	public void execute(Runnable runnable) {
		executor.execute(runnable);
	}
	
	/**
	 * @return name
	 *            線程池名稱 the String of pool name
	 */
	protected abstract String getThreadPoolName();
	
	/**
	 * @return corePoolSize
	 *            核心線程池大小 the number of threads to keep in the pool, even if
	 *            they are idle, unless {@code allowCoreThreadTimeOut} is set
	 */
	protected int corePoolSize(){
		return 5;
	}
	
	/**
	 * @return maximumPoolSize
	 *            最大線程池大小 the maximum number of threads to allow in the pool
	 */
	protected int maximumPoolSize(){
		return 10;
	}
	
	/**
	 * @return wattingCount
	 *            阻塞任務隊列數
	 */
	protected int wattingCount(){
		return 200000;
	}
	
	/**
	 * @return keepAliveTime
	 *            線程池中超過corePoolSize數目的空閒線程最大存活時間 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.
	 */
	protected long keepAliveTime(){
		return 10;
	}
}
複製代碼

子系統建立類 繼承ThreadPoolManager，配置參數信息 ViThreadPoolManager.java

package com.tool.me.thread;

/**
 * 當前類（子系統中定義的類）繼承 ThreadPoolManager 類，設置相關參數
 */
public class ViThreadPoolManager extends ThreadPoolManager{
	private static ThreadPoolManager threadPool  = null;
	
	public synchronized static ThreadPoolManager getInstance() {
		if(threadPool == null) {
			threadPool = new ViThreadPoolManager();
		}
		return threadPool; 
	}
	
	@Override
	protected String getThreadPoolName() {
		return "com.tool.me.vi";
	}

	@Override
	protected int corePoolSize() {
		/**
		 * 代碼 設置返回值
		 */
		return 10;
	}

	@Override
	protected int maximumPoolSize() {
		/**
		 * 代碼 設置返回值
		 */
		return 20;
	}
}
複製代碼

使用線程池 main

public static void main(String[] args) {
		ViThreadPoolManager.getInstance().execute(new Runnable() {
			@Override
			public void run() {
				io();
			}
		});
	}
複製代碼