线程池

1、线程池概念

  多线程技术主要解决处理器单元内多个线程执行的问题，可以显著减少处理器单元的闲置时间，增加处理器单元的吞吐能力。假设一个服务器完成一项任务所需时间为：

>T1创建线程时间

>T2在线程中执行任务的时间

>T3销毁线程时间

  如果T1+T3远大于T2，则可以选择线程池以提高服务器性能。线程池正是关注如何缩短和调整T1、T3时间的技术，从而提高服务器程序的性能，把T1、T3安排在服务器启动、结束的时间段或者一些空闲的时间段，这样当客户发送请求时就不会有T1、T3的开销了。

 

2、基本组成

  1)、线程池管理器（ThreadPool）：用于创建并管理线程池，包括创建线程池，添加任务，销毁线程池。
  2)、工作线程（PoolWorker）：线程池中的线程，没有任务时处于等待状态，可以循环执行任务。
  3)、任务接口（Task）：每个任务必须实现的接口，以供工作线程调度任务的执行，它主要规定了任务的入口，任务执行后的收尾工作，任务的执行状态等。
  4)、任务队列（taskQueue）：存放未处理的任务，提供一种缓存机制。

 

3、核心思想

  外部任务可选择性实现Task接口，将其作为参数传递给线程管理器执行任务的方法中
  线程管理器中控制着任务队列的对象，在其执行任务的方法中，实际是将任务添加到队列中
  内部类工作线程，会扫描任务队列，空闲的工作线程会从队列中取出任务并执行

 

4、示例代码

1)、任务类

		public class Task implements Runnable {

			private static volatile int i = 1;

			// 执行任务
			@Override
			public void run() {
				System.out.println("任务" + (i++) + "已完成");
			}
		}

 

2)、线程池类，线程管理器：创建线程、执行任务、销毁线程、获取线程基本信息

		public final class ThreadPool {

			// 线程池中默认线程数
			private static int worker_num = 5;
			// 工作线程
			private WorkThread[] workThreads;
			// 完成任务数
			private static volatile int finished_task = 0;
			// 任务队列，List线程不安全
			private List<Runnable> taskQueue = new Vector<>();

			// 单例模式
			private static ThreadPool threadPool;

			private ThreadPool() {
				this(5);
			}

			// 创建线程池，worker_num为线程池中工作线程的个数
			private ThreadPool(int worker_num) {
				ThreadPool.worker_num = worker_num;
				workThreads = new WorkThread[worker_num];
				for (int i = 0; i < worker_num; i++) {
					workThreads[i] = new WorkThread();
					workThreads[i].start();// 开启线程
				}
			}

			public static ThreadPool createThreadPool() {
				return createThreadPool(ThreadPool.worker_num);
			}

			public static ThreadPool createThreadPool(int worker_num) {
				if (worker_num <= 0) {
					worker_num = ThreadPool.worker_num;
				}
				if (threadPool == null) {
					threadPool = new ThreadPool(worker_num);
				}
				return threadPool;
			}

			// 执行任务，只是把任务加入到队列中，具体执行由线程池管理器决定
			public void excute(Runnable task) {
				synchronized (taskQueue) {
					taskQueue.add(task);
					taskQueue.notify();
				}
			}

			// 批量执行，只是把任务加入到队列中，具体执行由线程池管理器决定
			public void excute(Runnable[] tasks) {
				synchronized (taskQueue) {
					for (Runnable task : tasks) {
						taskQueue.add(task);
						taskQueue.notify();
					}
				}
			}

			// 批量执行，只是把任务加入到队列中，具体执行由线程池管理器决定
			public void excute(List<Runnable> tasks) {
				synchronized (taskQueue) {
					for (Runnable task : tasks) {
						taskQueue.add(task);
						taskQueue.notify();
					}
				}
			}

			// 销毁线程池，保证所有任务都完成的情况下才销毁所有线程，否则等待任务完成才销毁
			public void destroy() {
				// 如果还有任务未完成，就先等待
				while (!taskQueue.isEmpty()) {
					try {
						Thread.sleep(100);
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
				// 工作线程停止，且置为null
				for (int i = 0; i < worker_num; i++) {
					workThreads[i].stopWorker();
					workThreads[i] = null;
				}
				threadPool = null;
				taskQueue.clear();
			}

			// 返回工作线程的个数
			public int getWorkThreadNumber() {
				return worker_num;
			}

			// 返回已完成任务的个数，实际上指出了任务队列的个数
			public int getFinishedTaskNumber() {
				return finished_task;
			}

			// 返回未处理的任务个数，即任务队列的长度
			public int getWaitTaskNumber() {
				return taskQueue.size();
			}

			@Override
			public String toString() {
				synchronized (taskQueue) {
					return "WorkThread number:" + worker_num + ",finished work number:"
							+ finished_task + ",wait task number:"
							+ getWaitTaskNumber();
				}
			}

			//内部类，工作线程
			private class WorkThread extends Thread {

				// 线程是否有效，用于控制线程的结束
				private boolean isRunning = true;

				// 如果任务队列不空，则取出任务执行；如果任务队列空，则等待
				@Override
				public void run() {
					Runnable r = null;

					while (isRunning) {
						synchronized (taskQueue) {
							while (isRunning && taskQueue.isEmpty()) {// 队列为空
								try {
									taskQueue.wait(200);
								} catch (InterruptedException e) {
									e.printStackTrace();
								}
							}
							if (!taskQueue.isEmpty()) {
								r = taskQueue.remove(0);
							}
							if (r != null) {
								r.run();// 执行任务
							}

							finished_task++;
							r = null;
						}
					}

				}

				// 停止工作，让程序自然执行完run方法，自然结束
				public void stopWorker() {
					isRunning = false;
				}
			}
		}

 

3)、客户端测试

		public class Client {

			public static void main(String[] args) {

				// 创建3个线程的线程池
				ThreadPool t = ThreadPool.createThreadPool(3);
				t.excute(new Runnable[] { new Task(), new Task(), new Task() });
				t.excute(new Runnable[] { new Task(), new Task(), new Task() });
				t.excute(new Runnable[] { new Task(), new Task(), new Task() });
				t.excute(new Runnable[] { new Task(), new Task(), new Task() });

				// 所有线程都执行完毕后才destroy
				t.destroy();
				System.out.println(t);
			}
		}

 

评论

1 楼 niunianss 2015-07-31  

workthread  synchronized (taskQueue)粒度是不是太大，必须一个worker工作完成后，下一个worker才有可能工作，造成worker串行工作，个人见解，只为交流