Python 线程池
import Queue, threading, sys
from threading import Thread
import time,urllib
# working thread
class Worker(Thread):
worker_count = 0
def __init__( self, workQueue, resultQueue, timeout = 0, **kwds):
Thread.__init__( self, **kwds )
self.id = Worker.worker_count
Worker.worker_count += 1
self.setDaemon( True )
self.workQueue = workQueue
self.resultQueue = resultQueue
self.timeout = timeout
self.start( )
def run( self ):
''' the get-some-work, do-some-work main loop of worker threads '''
while True:
try:
callable, args, kwds = self.workQueue.get(timeout=self.timeout)
res = callable(*args, **kwds)
print "worker[%2d]: %s" % (self.id, str(res) )
self.resultQueue.put( res )
except Queue.Empty:
break
except :
print 'worker[%2d]' % self.id, sys.exc_info()[:2]
class WorkerManager:
def __init__( self, num_of_workers=10, timeout = 1):
self.workQueue = Queue.Queue()
self.resultQueue = Queue.Queue()
self.workers = []
self.timeout = timeout
self._recruitThreads( num
相关文档：
我先给一个初步的表格吧，大家如果有什么意见，或有补充，欢迎提出。有些我没有用过，先不写了。
以下是我使用过的python IDE:
除了PythonWin, VisualPython只支持Windows，其它都至少支持Win/Linux/Mac。
各项含义：
自动补全：变量/函数名打到一半时，提示可能的完整的变量/函数名� ......
import os
import unittest # 包含单元测试模块
import sqlite3 as sqlite # 包含sqlite3模块
def get_db_path():
return "sqlite_testdb"
class TransactionTests(unittest.TestCase): # 单元测试第一步: 由TestCase派生类
def setUp(self): # 单元测试环境配置
......
在python中如何创建一个线程对象
如果你要创建一个线程对象，很简单，只要你的类继承threading.Thread，然后在__init__里首先调用threading.Thread的__init__方法即可
import threading
class mythread(threading.Thread):
def __init__(self, threadname):
& ......
在项目里面一个解析文本的工具里面用到了这个命令来赋值，开始一直知道意思，呵呵 查了下，找到方法如下：
exec
语句用来执行储存在字符串或文件中的Python语句。例如，我们可以在运行时生成一个包含Python代码的字符串，然后使用
exec
语句执行这些语句。下面是一个简单的例子。
> ......
