版本一

#!/usr/bin/env??python
#?--*--coding:utf-8?--*--
import?Queue
import?threading


class?ThreadPool(object):
#創(chuàng)建類
????def?__init__(self,?max_num=20):#進(jìn)程函數(shù)，默認(rèn)最大20個(gè)進(jìn)程
????????self.queue?=?Queue.Queue(max_num)#生成進(jìn)程
????????for?i?in?xrange(max_num):#循環(huán)進(jìn)程
????????????self.queue.put(threading.Thread)#上傳進(jìn)程

????def?get_thread(self):#下載進(jìn)程函數(shù)
????????return?self.queue.get()

????def?add_thread(self):#生成進(jìn)程函數(shù)
????????self.queue.put(threading.Thread)


pool?=?ThreadPool(10)#執(zhí)行類，并傳默認(rèn)進(jìn)程數(shù)值

def?func(arg,?p):#打印進(jìn)程
????print?arg
????import?time
????time.sleep(2)#間隔2秒
????p.add_thread()


for?i?in?xrange(30):#循環(huán)進(jìn)程
????thread?=?pool.get_thread()
????t?=?thread(target=func,?args=(i,?pool))#傳值到func函數(shù)，并且執(zhí)行
????t.start()

?版本二

from?Queue?import?Queue
import?contextlib
import?threading
?
WorkerStop?=?object()
?
?
class?ThreadPool:
?
????workers?=?0
?
????threadFactory?=?threading.Thread
????currentThread?=?staticmethod(threading.currentThread)
?
????def?__init__(self,?maxthreads=20,?name=None):
?
????????self.q?=?Queue(0)
????????self.max?=?maxthreads
????????self.name?=?name
????????self.waiters?=?[]
????????self.working?=?[]
?
????def?start(self):
????????while?self.workers?<?min(self.max,?self.q.qsize()+len(self.working)):
????????????self.startAWorker()
?
????def?startAWorker(self):
????????self.workers?+=?1
????????name?=?"PoolThread-%s-%s"?%?(self.name?or?id(self),?self.workers)
????????newThread?=?self.threadFactory(target=self._worker,?name=name)
????????newThread.start()
?
????def?callInThread(self,?func,?*args,?**kw):
????????self.callInThreadWithCallback(None,?func,?*args,?**kw)
?
????def?callInThreadWithCallback(self,?onResult,?func,?*args,?**kw):
????????o?=?(func,?args,?kw,?onResult)
????????self.q.put(o)
?
?
????@contextlib.contextmanager
????def?_workerState(self,?stateList,?workerThread):
????????stateList.append(workerThread)
????????try:
????????????yield
????????finally:
????????????stateList.remove(workerThread)
?
????def?_worker(self):
????????ct?=?self.currentThread()
????????o?=?self.q.get()
????????while?o?is?not?WorkerStop:
????????????with?self._workerState(self.working,?ct):
????????????????function,?args,?kwargs,?onResult?=?o
????????????????del?o
????????????????try:
????????????????????result?=?function(*args,?**kwargs)
????????????????????success?=?True
????????????????except:
????????????????????success?=?False
????????????????????if?onResult?is?None:
????????????????????????pass
?
????????????????????else:
????????????????????????pass
?
????????????????del?function,?args,?kwargs
?
????????????????if?onResult?is?not?None:
????????????????????try:
????????????????????????onResult(success,?result)
????????????????????except:
????????????????????????#context.call(ctx,?log.err)
????????????????????????pass
?
????????????????del?onResult,?result
?
????????????with?self._workerState(self.waiters,?ct):
????????????????o?=?self.q.get()
?
????def?stop(self):
????????while?self.workers:
????????????self.q.put(WorkerStop)
????????????self.workers?-=?1
?
?
"""
def?show(arg):
????import?time
????time.sleep(1)
????print?arg
?
?
pool?=?ThreadPool(20)
?
for?i?in?range(500):
????pool.callInThread(show,?i)
?
pool.start()
pool.stop()
"""

?

轉(zhuǎn)載于:https://www.cnblogs.com/wulaoer/p/5122472.html

總結(jié)

以上是生活随笔為你收集整理的Python之路：线程池的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。