一个可中断的线程池python实现

我有一个需求，通过多线程测试python memcached的性能:

class TestThread(threading.Thread):
	def __init__(self):
		threading.Thread.__init__(self)

	def run(self):
		while True:
			print mc.get('test')

for i in xrange(8):
	t = TestThread()
	t.start()

 

但是坑爹的是，怎么优雅的让他停止呢？我实现了一个Executor

 

# -*- coding: utf-8 -*-
import re
import urllib2
import json
import threading
import Queue
import os
import time
from time import sleep
from threading import Lock
from signal import signal,SIGTERM,SIGINT,SIGQUIT


class Executor:
	def __init__(self,size):	
		self.queue = Queue.Queue()
		self.tasks = []
		self.running = True

		for i  in range(size):
			t = Task(self.queue)
			t.setDaemon(True)
			t.start()
			self.tasks.append(t)

		self._signal()

	def _signal(self):
		signal(SIGTERM,self._exit)
		signal(SIGINT,self._exit)
		signal(SIGQUIT,self._exit)


	def _exit(self,a=None,b=None):
		print '_exit'
		self.cancel()
	
	def cancel(self):
		for task in self.tasks:
		    while not task.cancel():
			#print 'try again %s' %task.getName()
			#sleep(0.1)
			pass

		self.running = False
		self.onCancel()
				
	def submit(self,call):
		self.queue.put(call)	
	
	def join(self):
		#self.queue.join() queue.join()会阻塞,所以不用
		while self.running and not self.queue.empty():
			sleep(0.1)
			if self.hasattr('cancelTrigger') and self.cancelTrigger():
				self.cancel()

	def setCancelTrigger(self,cancelTrigger):
		self.cancelTrigger = cancelTrigger

	def setOnCancel(self,onCancel):
		self.onCancel = onCancel
		#print self.onCancel

class Task(threading.Thread):

	def __init__(self,queue):
		threading.Thread.__init__(self)
		self.queue = queue
		self.running = True
		self.canceled = False

	def cancel(self):
		#self.canceled=True
		self.running=False
		return self.isCanceled()

	def isCanceled(self):
		return self.running==False
	
	def run(self):
		while self.running:
			call = self.queue.get()
			call.run()
			self.queue.task_done()
			if self.canceled:
			    self.running = False

 

如何使用：

class Checker():

	def run(self):
		print mc.get('test')

executor = Executor(10)
for x in xrange(100):
     executor.submit(Checker())
executor.join()