JAVA线程池例子

JAVA线程池例子

· 用途及用法

 

 

网络请求通常有两种形式：第一种，请求不是很频繁，而且每次连接后会保持相当一段时间来读数据或者写数据，最后断开，如文件下载，网络流媒体等。另一种形式是请求频繁，但是连接上以后读/写很少量的数据就断开连接。考虑到服务的并发问题，如果每个请求来到以后服务都为它启动一个线程，那么这对服务的资源可能会造成很大的浪费，特别是第二种情况。因为通常情况下，创建线程是需要一定的耗时的，设这个时间为T1，而连接后读/写服务的时间为T2，当T1>>T2时，我们就应当考虑一种策略或者机制来控制，使得服务对于第二种请求方式也能在较低的功耗下完成。

通常，我们可以用线程池来解决这个问题，首先，在服务启动的时候，我们可以启动好几个线程，并用一个容器(如线程池)来管理这些线程。当请求到来时，可以从池中去一个线程出来，执行任务(通常是对请求的响应)，当任务结束后，再将这个线程放入池中备用；如果请求到来而池中没有空闲的线程，该请求需要排队等候。最后，当服务关闭时销毁该池即可。

· 结构

线程池中通常由这样几个概念(接口)组成：

1 线程池(Thread pool )，池是一个容器，容器中有很多个执行器，每一个执行器是一个线程。当然，这个容器的实现，可以是链表，可以是数组等等，不需要关心，需要关心的是，池必须提供一个可以从中取出执行器 的方法，可能还需要一个池中现有活动线程数方法，销毁池的方法等。

2 执行器(Executor )，每个执行器是一个线程，每个执行器可以执行一个任务 ，任务是做什么，此时还不很明确，它需要提供任务的setter/getter方法，并且作为一个线程，他可以独立运行，执行器执行完自身后，需要将自身放入池中。

3 任务(Task )，任务是每个线程具体要做的事，如资源下载，播放flash片段，打印一段文字到控制台等等，它本身不能执行，而需要将自身交给执行器。

整个池的机制和结构就是这样，当然，需要一个调度者(scheduler)来协调主线程和池的关系。结构，或者接口的目的是为了让我们从细节中解脱出来，从一个比较抽象的层次来描述系统，这样的好处是简单，而且设计出来的框架比较通用，可以适应很多相近相似的情况。由于Task具体干什么我们不知道，所以它几乎可以干任何适应于上边总结的网络连接的第二种情况(T1>>T2)。

· 类的结构图

虽然为一个简单的实现设计一个标准的UML视图是不太现实的，但是这是一种受鼓励的做法，至少应该用铅笔在草纸上画出相关的视图，这样可以帮助以后的维护和更高级的扩展。

· 线程池的简单实现

实现可以是通过多种语言的，我们在此选择面向对象的JAVA，而如果你使用C的话，也没有问题，问题在上一小节已经描述清楚，语言是不重要的。

池是一个容器，我们考虑使用java.util.LinkedList类(可能由于它的长度是可变的，而且不需要我们使用者来考虑)，也就是说，池需要维护一个链表。

 

Java代码 

4 public   interface  Pool { //池接口    

5     Executor getExecutor();   

6     void destroy();   

7 }  

[java]view plaincopyprint?

8 public interface Pool {//池接口   

9     Executor getExecutor();  

10     void destroy();  

11 }  

  

 

Java代码 

12 public   interface  Executor { //执行器接口    

13     void setTask(Task task);   

14     Task getTask();   

15     void startTask();   

16 }  

[java]view plaincopyprint?

17 public interface Executor {//执行器接口   

18     void setTask(Task task);  

19     Task getTask();  

20     void startTask();  

21 }  

  

 

鉴于执行器是池中的对象，而且外部没有必要知道其细节，我们考虑将Executor接口的实现做为Pool接口的实现的内部类。这样做的另一个好处是，更便于池的管理。

 

Java代码 

22 import  java.util.LinkedList;    

23 import java.util.Properties;   

24   

25 import redesigned.utils.PropReader;   

26   

27 public class ThreadPool implements Pool{   

28     private boolean isShut;   

29     private LinkedList pool;   

30     private static Properties prop = PropReader.getProperties("webconfig.properties");   

31     private int size = Integer.parseInt(prop.getProperty("threadsperpage", "3"));   

32     public ThreadPool(){   

33         // read configuration and set the   

34         // content of pool by objects of Executor   

35         isShut = false;//set the status of pool to active   

36         pool = new LinkedList();   

37         for(int i = 0; i < size; i++){   

38             Executor executor = new ExecutorImpl();//new a executor thread   

39             pool.add(executor);//add it to pool   

40             ((ExecutorImpl)executor).start();//start it   

41         }   

42     }   

43     public void destroy() {//Destroy   

44         synchronized(pool){   

45             isShut = true;//set the status of pool to inactive   

46             pool.notifyAll();//notify all listener.   

47             pool.clear();//clear the list of threads   

48         }   

49     }   

50   

51     public Executor getExecutor(){   

52         Executor ret = null;   

53         synchronized(pool){//return if any.   

54             if(pool.size() > 0){   

55                 ret = (Executor)pool.removeFirst();   

56             }else{   

57                 try {   

58                     pool.wait();   

59                 } catch (InterruptedException e) {   

60                     e.printStackTrace();   

61                 }   

62                 ret = (Executor)pool.removeFirst();   

63             }   

64         }   

65         return ret;   

66     }   

67   

68 Executor接口的实现作为ThreadPool的内部类   

69     private class ExecutorImpl extends Thread implements Executor{   

70         private Task task;   

71         private Object lock = new Object();   

72         //private boolean loop = true;   

73         public ExecutorImpl(){}   

74         public Task getTask() {   

75             return this.task;   

76         }   

77   

78         public void setTask(Task task) {   

79             this.task = task;   

80         }   

81         public void startTask(){   

82             //System.out.println("start here");   

83             synchronized(lock){   

84                 lock.notify();   

85             }   

86         }   

87         public void run(){   

88             //get a task if any   

89             //then run it   

90             //then put self to pool   

91             while(!isShut){   

92                 synchronized(lock){   

93                     try {   

94                         lock.wait();//wait for resource   

95                     } catch (InterruptedException e) {   

96                         e.printStackTrace();   

97                     }   

98                 }   

99                 getTask().execute();//execute the task   

100                 synchronized(pool){//put it self to the pool when finish the task   

101                     pool.addFirst(ExecutorImpl.this);   

102                     pool.notifyAll();   

103                 }   

104             }   

105         }   

106     }   

107 }   

[java]view plaincopyprint?

108 import java.util.LinkedList;  

109 import java.util.Properties;  

110   

111 import redesigned.utils.PropReader;  

112   

113 public class ThreadPool implements Pool{  

114     private boolean isShut;  

115     private LinkedList pool;  

116     private static Properties prop = PropReader.getProperties("webconfig.properties");  

117     private int size = Integer.parseInt(prop.getProperty("threadsperpage", "3"));  

118     public ThreadPool(){  

119         // read configuration and set the   

120         // content of pool by objects of Executor   

121         isShut = false;//set the status of pool to active   

122         pool = new LinkedList();  

123         for(int i = 0; i < size; i++){  

124             Executor executor = new ExecutorImpl();//new a executor thread   

125             pool.add(executor);//add it to pool   

126             ((ExecutorImpl)executor).start();//start it   

127         }  

128     }  

129     public void destroy() {//Destroy   

130         synchronized(pool){  

131             isShut = true;//set the status of pool to inactive   

132             pool.notifyAll();//notify all listener.   

133             pool.clear();//clear the list of threads   

134         }  

135     }  

136   

137     public Executor getExecutor(){  

138         Executor ret = null;  

139         synchronized(pool){//return if any.   

140             if(pool.size() > 0){  

141                 ret = (Executor)pool.removeFirst();  

142             }else{  

143                 try {  

144                     pool.wait();  

145                 } catch (InterruptedException e) {  

146                     e.printStackTrace();  

147                 }  

148                 ret = (Executor)pool.removeFirst();  

149             }  

150         }  

151         return ret;  

152     }  

153   

154 Executor接口的实现作为ThreadPool的内部类  

155     private class ExecutorImpl extends Thread implements Executor{  

156         private Task task;  

157         private Object lock = new Object();  

158         //private boolean loop = true;   

159         public ExecutorImpl(){}  

160         public Task getTask() {  

161             return this.task;  

162         }  

163   

164         public void setTask(Task task) {  

165             this.task = task;  

166         }  

167         public void startTask(){  

168             //System.out.println("start here");   

169             synchronized(lock){  

170                 lock.notify();  

171             }  

172         }  

173         public void run(){  

174             //get a task if any   

175             //then run it   

176             //then put self to pool   

177             while(!isShut){  

178                 synchronized(lock){  

179                     try {  

180                         lock.wait();//wait for resource   

181                     } catch (InterruptedException e) {  

182                         e.printStackTrace();  

183                     }  

184                 }  

185                 getTask().execute();//execute the task   

186                 synchronized(pool){//put it self to the pool when finish the task   

187                     pool.addFirst(ExecutorImpl.this);  

188                     pool.notifyAll();  

189                 }  

190             }  

191         }  

192     }  

193 }   

 

 

好了，池设计好了，再来看看任务(Task)的接口和实现

Java代码 

194 public   interface  Task { //这个接口也比较简单，可以执行，可以取到执行结果    

195     void execute();   

196     byte[] getResult();   

197 }  

[java]view plaincopyprint?

198 public interface Task {//这个接口也比较简单，可以执行，可以取到执行结果   

199     void execute();  

200     byte[] getResult();  

201 }  

  

 

Task的实现可以是多种多样的，下边的例子是一个加载资源的Task.使用方式

 

 

Java代码 

202 Pool pool =  new  ThreadPool(); // new a ThreadPool    

203 //load resources on each page, and start #s of thread.   

204 for(int i = 0; i < resourceList.size();i++){   

205     Executor executor = pool.getExecutor(); // get Executor form pool   

206     Task resourceLoader = new ResourceLoader((String)resourceList.get(i));   

207     executor.setTask(resourceLoader); // set the task to executor   

208     executor.startTask(); // try to start the executor.   

209 }   

210   

211 //wait while all task are done, the destroy the pool.   

212 pool.destroy();   

[java]view plaincopyprint?

213 Pool pool = new ThreadPool();// new a ThreadPool   

214 //load resources on each page, and start #s of thread.   

215 for(int i = 0; i < resourceList.size();i++){  

216     Executor executor = pool.getExecutor(); // get Executor form pool   

217     Task resourceLoader = new ResourceLoader((String)resourceList.get(i));  

218     executor.setTask(resourceLoader); // set the task to executor   

219     executor.startTask(); // try to start the executor.   

220 }  

221   

222 //wait while all task are done, the destroy the pool.   

223 pool.destroy();   

 

· 优势，或者适用范围

224 在并发时，需要被并发的线程不需要知道自己什么时候需要被启动，它子需要考虑这样一种情况：它自己从一个地方取出来一个执行器，然后把任务交给执行器，然后等待执行器结束即可，他关心的是自己所需要干的事，或者自己负责的事。这样，大家都简单，因为只需要做好自己的事情就好了。面向对象的一个秘诀为：永远相信合作者，使用别人的接口而不是自己去实现所有的接口。

225 这种T1>>T2的请求方式在网络中固然是常见的，在实际问题中同样是常见的。因此，掌握这种模式可能会对我们以后的程序设计提供方便和好处。

· 小结

同步问题： 同步在线程的并发中意义非常之大，对临界资源的控制是并发时最关键的地方。如在线程池中，当池中没有空闲的线程时，新来的请求就必须等待，而一旦一个Task运行结束后，一方面将自己放入池中，一方面需要通知等待在pool中的其他线程。每一个执行器线程，一开始启动，则进入等待状态，此时不会消耗CPU资源。而当在外部调用执行器的startTask()方法，即可通知线程从等待状态中醒来，去出Task，执行之，将执行器本身放入池中，然后继续等待。

 

当然，实现的策略是可以多种多样的，但是问题的本质已经在第二小节结构 很明确的被定义了。

 

最近开始学习JAVA，同时也开始熟悉面向对象的思维方式，这篇日志，一来作为一个备忘，二来可以对可能需要的人提供帮助。以上可以算作是小结。 

<!--EndFragment-->