文档章节

一个简单的ThreadPool实现,以及基于线程池的简单的Web服务器

lee123lee
 lee123lee
发布于 2016/09/05 12:00
字数 862
阅读 86
收藏 3

 一个简单的线程池

package com;

public interface ThreadPool<Job extends Runnable> {
	//执行一个JOB
	void execute(Job job);
	//关闭线程池
	void shutdown();
	//增加工作者线程
	void addWorkers(int num);
	//减少工作者线程
	void removeWorker(int num);
	//得到正在等待执行的任务数量
	int getJobSize();
}
package com;

import java.util.ArrayList;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;

public class DefaultThreadPool<Job extends Runnable> implements ThreadPool<Job> {
	//线程池最大工作者限制数
	private static final int MAX_WORKER_NUM = 10;
	//线程池最小工作者限制数
	private static final int MIN_WORKER_NUM = 1;
	//线程池默认工作者数量
	private static final int DEFAULT_WORKER_NUM = 5;
	//工作列表
	private final LinkedList<Job> jobs = new LinkedList<>();
	
	//工作者线程数量
	private int workerNum = DEFAULT_WORKER_NUM;
	//线程编号生成
	private AtomicLong threadNum = new AtomicLong();
	//工作者列表
	private final List<Worker> workers = Collections.synchronizedList(new ArrayList<Worker>());
	
	public DefaultThreadPool() {
		initiallzeWorkers(DEFAULT_WORKER_NUM);
	}
	
	public DefaultThreadPool(int num) {
		workerNum = num > MAX_WORKER_NUM ? MAX_WORKER_NUM : num < MIN_WORKER_NUM ? MIN_WORKER_NUM : num;
		initiallzeWorkers(num);
	}
	
	
	
	@Override
	public void execute(Job job) {
		if(job != null){
			//添加一个工作,然后进行通知
			synchronized (jobs) {
				jobs.addLast(job);
				jobs.notify();
			}
		}
	}

	@Override
	public void shutdown() {
		for(Worker worker:workers){
			worker.shutdown();
		}
	}

	@Override
	public void addWorkers(int num) {
		synchronized (jobs) {
			//限制新增的worker数量不能超过最大值
			if(num + this.workerNum > MAX_WORKER_NUM){
				num = MAX_WORKER_NUM-this.workerNum;
			}
			initiallzeWorkers(num);
			this.workerNum += num;
		}
	}

	@Override
	public void removeWorker(int num) {
		synchronized (jobs) {
			if(num >= this.workerNum){
				throw new IllegalArgumentException("beyond workNum");
			}
			int count = 0;
			while(count < num){
				Worker worker = workers.get(count);
				if(workers.remove(worker)){
					worker.shutdown();
					count++;
				}
			}
			this.workerNum -= count;
		}
	}

	@Override
	public int getJobSize() {
		return jobs.size();
	}
	
	//初始化工作线程工作者
	private void initiallzeWorkers(int num){
		for(int i = 0; i < num; i++){
			Worker worker = new Worker();
			workers.add(worker);
			Thread thread = new Thread(worker,"ThreadPool-Worker-"+threadNum.incrementAndGet());
			thread.start();
		}
	}
	
	
	//工作者,消费任务
	class Worker implements Runnable{

		private volatile boolean running = true;
		
		@Override
		public void run() {
			while(running){
				Job job  = null;
				synchronized (jobs) {
					while(jobs.isEmpty()){
						try {
							jobs.wait();
						} catch (InterruptedException e) {
							//感知外部对workerThread的中断返回
							Thread.currentThread().interrupt();
							return;
						}
					}
					//取出job
					job = jobs.removeFirst();
				}
				if(job != null){
					try{
						job.run();
					}catch(Exception e){
						//忽略Job执行中的Exception
					}
				}
			}
		}
		
		public void shutdown(){
			running = false;
		}
	}

}

从线程池的实现中可以看出,当客户端调用  execute(Job) 方法时候,会不断的向任务列表 jobs 添加 job,每个工作者线程不断的从jobs中取出一个job进行执行,当jobs为空时候,工作者线程进入等待状态。

一个基于线程池技术的简单Web服务器

package com;

import java.io.BufferedReader;
import java.io.ByteArrayOutputStream;
import java.io.Closeable;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.ServerSocket;
import java.net.Socket;

public class SimpleHttpServer {

	//处理 HttpRequest的线程池
	static ThreadPool<HttpRequestHandler> threadPool = new DefaultThreadPool<HttpRequestHandler>(1);
	
	//SimpleHttpServer的根路径
	static String basePath;
	static ServerSocket serverSocket;
	static int port = 8080;

	public static void setPort(int port){
		if(port>0){
			SimpleHttpServer.port = port;
		}
	}
	public static void setBasePath(String basePath){
		if(basePath != null && new File(basePath).exists() && new File(basePath).isDirectory()){
			SimpleHttpServer.basePath = basePath;
		}
	}
	
	public static void start() throws IOException{
		serverSocket = new ServerSocket(port);
		Socket socket;
		while((socket = serverSocket.accept())!= null){
			threadPool.execute(new HttpRequestHandler(socket));
		}
		serverSocket.close();
	}
	
	public static void main(String[] args) {
		try {
			
			start();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	
	static class HttpRequestHandler implements Runnable{

		private Socket socket;
		
		public HttpRequestHandler(Socket socket) {
			this.socket = socket;
		}
		
		@Override
		public void run() {
			String line = null;
			BufferedReader br = null;
			BufferedReader reader = null;
			PrintWriter out = null;
			InputStream in = null;
			
			try {
				reader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
				String header = reader.readLine();
				//相对路径计算绝对路径
				String filePath = basePath + header.split(" ")[1];
				out = new PrintWriter(socket.getOutputStream());
				
				//后缀为jpg或者ico,则读取资源输出
				if(filePath.endsWith("jpg") || filePath.endsWith("ico")){
					in = new FileInputStream(filePath);
					ByteArrayOutputStream baos = new ByteArrayOutputStream();
					int i = 0;
					while( (i = in.read()) != -1 ){
						baos.write(i);
					}
					byte[] array = baos.toByteArray();
					out.println("HTTP/1.1 200 OK");
					out.println("Server: TOMCAT");
					out.println("Content-Type: image/jpeg");
					out.println("Content-Length: "+array.length);
					out.println("");
					socket.getOutputStream().write(array, 0, array.length);
				}else{
					br = new BufferedReader(new InputStreamReader(new FileInputStream(filePath)));
					out.println("HTTP/1.1 200 OK");
					out.println("Server: TOMCAT");
					out.println("Content-Type: text/html; charset=UTF-8");
					out.println("");
					while((line = br.readLine())!=null){
						out.println(line);
					}
				}
				out.flush();
			} catch (IOException e) {
				out.println("HTTP/1.1 500 OK");
				out.println("");
				out.flush();
			}finally {
				close(br,in,reader,out,socket);
			}
		}
		
		private static void close(Closeable... closeables){
			if(closeables != null){
				for (Closeable closeable : closeables) {
					if(closeable!= null){
						try {
							closeable.close();
						} catch (IOException e) {
							e.printStackTrace();
						}
					}
				}
			}
		}
		
	}
}

 

© 著作权归作者所有

lee123lee
粉丝 51
博文 144
码字总数 122159
作品 1
闵行
高级程序员
私信 提问
boost之ThreadPool

threadpool是基于boost库实现的一个线程池子库,但线程池实现起来不是很复杂。我们从threadpool中又能学到什么东西呢? 它是基于boost库实现的,如果大家对boost库有兴趣,看看一个简单的实现...

mjrao
2015/09/04
483
0
Asp.Net Core 轻松学-多线程之Task快速上手

前言     Task是从 .NET Framework 4 开始引入的一项基于队列的异步任务(TAP)模式,从 .NET Framework 4.5 开始,任何使用 async/await 进行修饰的方法,都会被认为是一个异步方法;实...

Ron.liang
2018/12/31
0
0
Linux下通用线程池的创建与使用

Linux下通用线程池的创建与使用 本文给出了一个通用的线程池框架,该框架将与线程执行相关的任务进行了高层次的抽象,使之与具体的执行任务无关。另外该线程池具有动态伸缩性,它能根据执行任...

横云断岭
2010/10/06
0
0
java多线程总结五:线程池的原理及实现

1、线程池简介: 多线程技术主要解决处理器单元内多个线程执行的问题,它可以显著减少处理器单元的闲置时间,增加处理器单元的吞吐能力。 假设一个服务器完成一项任务所需时间为:T1 创建线程...

陆朋
2013/05/28
0
2
Rust 并发编程 - Thread Pool

在并发编程领域,一个非常让程序员兴奋,感到有成就感的事情就是做性能优化,譬如发现某个线程成为了单点瓶颈,然后上多线程。 提到了上多线程,那自然就会引入 thread pool,也就是我们通常...

siddontang
03/30
0
0

没有更多内容

加载失败,请刷新页面

加载更多

Java中print、printf、println的区别

printf主要是继承了C语言的printf的一些特性,可以进行格式化输出 print就是一般的标准输出,但是不换行 println和print基本没什么差别,就是最后会换行

hellation_
12分钟前
0
0
spring在静态类中注入bean的的解释

@Componentpublic class ModelMapper {@AutoWiredprivate static AssignmentManager assignmentManager;public static void add(){a+b;}} 静态方法是属于类的,普通方法才属于...

无知的小狼
13分钟前
0
0
分而治之-归并排序

如果有1个数组,数组的左半部分和右半部分都已经排好序,如何将该数组合成1个有序的数组? 开辟1个同样大小的临时空间辅助我们完成归并过程,如下图 k:表示归并过程中,当前需要替换的原数组...

万山红遍
27分钟前
0
0
Linux修改时区的正确方法【修改时间,需要修改软连接,靠谱】

CentOS和Ubuntu的时区文件是/etc/localtime,但是在CentOS7以后localtime以及变成了一个链接文件 [root@centos7 ~]# ll /etc/localtime lrwxrwxrwx 1 root root 33 Oct 12 11:01 /etc/loca......

Airship
今天
1
0
《Netkiller Spring Cloud 手札》之 Master / Slave 主从数据库数据源配置

5.19.1. Master / Slave 主从数据库数据源配置 5.19.1.1. application.properties spring.datasource.master.driverClassName = com.mysql.cj.jdbc.Driverspring.datasource.master.url=j......

netkiller-
今天
50
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部