1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
|
"""
一个基于thread和queue的线程池，以任务为队列元素，动态创建线程，重复利用线程，
通过close和terminate方法关闭线程池。
"""
try:
import queue
except:
import Queue
import threading
import contextlib
import time
StopEvent = object()
StoreCallBack={}
def callback(status, result):
"""
根据需要进行的回调函数，默认不执行。
:param status: action函数的执行状态
:param result: action函数的返回值
:return:
"""
print result
StoreCallBack[result[u'task']]=result["result"]
return status,result
def action(thread_name, arg):
"""
真实的任务定义在这个函数里
:param thread_name: 执行该方法的线程名
:param arg: 该函数需要的参数
:return:
"""
return {u"result":u"dsaf",u"task":arg}
class ThreadPool:
def __init__(self, max_num, max_task_num=None):
"""
初始化线程池
:param max_num: 线程池最大线程数量
:param max_task_num: 任务队列长度
"""
if max_task_num:
try:
self.q = queue.Queue(max_task_num)
except:
self.q = Queue.Queue(max_task_num)
else:
try:
self.q = queue.Queue()
except:
self.q = Queue.Queue()
self.max_num = max_num
self.cancel = False
self.terminal = False
self.generate_list = []
self.free_list = []
def put(self, func, args, callback=None):
"""
往任务队列里放入一个任务
:param func: 任务函数
:param args: 任务函数所需参数
:param callback: 任务执行失败或成功后执行的回调函数，回调函数有两个参数
1、任务函数执行状态；2、任务函数返回值（默认为None，即：不执行回调函数）
:return: 如果线程池已经终止，则返回True否则None
"""
if self.cancel:
return
if len(self.free_list) == 0 and len(self.generate_list) < self.max_num:
self.generate_thread()
w = (func, args, callback,)
self.q.put(w)
def generate_thread(self):
"""
创建一个线程
"""
t = threading.Thread(target=self.call)
t.start()
def call(self):
"""
循环去获取任务函数并执行任务函数。在正常情况下，每个线程都保存生存状态， 直到获取线程终止的flag。
"""
current_thread = threading.currentThread().getName()
self.generate_list.append(current_thread)
event = self.q.get()
while event != StopEvent:
func, arguments, callback = event
try:
result = func(current_thread, *arguments)
success = True
except Exception as e:
result = None
success = False
if callback is not None:
try:
callback(success, result)
except Exception as e:
pass
with self.worker_state(self.free_list, current_thread):
if self.terminal:
event = StopEvent
else:
event = self.q.get()
else:
self.generate_list.remove(current_thread)
def close(self):
"""
执行完所有的任务后，让所有线程都停止的方法
"""
self.cancel = True
full_size = len(self.generate_list)
while full_size:
self.q.put(StopEvent)
full_size -= 1
def terminate(self):
"""
在任务执行过程中，终止线程，提前退出。
"""
self.terminal = True
while self.generate_list:
self.q.put(StopEvent)
@contextlib.contextmanager
def worker_state(self, state_list, worker_thread):
"""
用于记录空闲的线程，或从空闲列表中取出线程处理任务
"""
state_list.append(worker_thread)
try:
yield
finally:
state_list.remove(worker_thread)
class Rtet:
def __init__(self,a,b,c):
self.a = a
self.b = b
self.c = c
def ab(self,thread_name, arg):
print(thread_name)
return self.a+self.b
def ac(self,thread_name):
print(thread_name)
return self.c*self.a
def threadPoolCore():
pool = ThreadPool(5)
t = Rtet(23, 12, 34)
pool.put(action, (u"test",), callback)
pool.put(action, (u"test2",), callback)
pool.put(action, (u"test3",), callback)
time.sleep(3)
print("-" * 50)
print(u"\033[32;0m任务停止之前线程池中有%s个线程，空闲的线程有%s个！\033[0m"
% (len(pool.generate_list), len(pool.free_list)))
pool.close()
print(u"任务执行完毕，正常退出！")
return StoreCallBack
threadPoolCore()
|
