Python中的并发编程实例
一、简介
我们将一个正在运行的程序称为进程。每个进程都有它自己的系统状态，包含内存状态、打开文件列表、追踪指令执行情况的程序指针以及一个保存局部变量的调用栈。通常情况下，一个进程依照一个单序列控制流顺序执行，这个控制流被称为该进程的主线程。在任何给定的时刻，一个程序只做一件事情。
一个程序可以通过Python库函数中的os或subprocess模块创建新进程(例如os.fork()或是subprocess.Popen())。然而，这些被称为子进程的进程却是独立运行的，它们有各自独立的系统状态以及主线程。因为进程之间是相互独立的，因此它们同原有的进程并发执行。这是指原进程可以在创建子进程后去执行其它工作。
虽然进程之间是相互独立的，但是它们能够通过名为进程间通信(IPC)的机制进行相互通信。一个典型的模式是基于消息传递，可以将其简单地理解为一个纯字节的缓冲区，而send()或recv()操作原语可以通过诸如管道(pipe)或是网络套接字(networksocket)等I/O通道传输或接收消息。还有一些IPC模式可以通过内存映射(memory-mapped)机制完成(例如mmap模块)，通过内存映射，进程可以在内存中创建共享区域，而对这些区域的修改对所有的进程可见。
多进程能够被用于需要同时执行多个任务的场景，由不同的进程负责任务的不同部分。然而，另一种将工作细分到任务的方法是使用线程。同进程类似，线程也有其自己的控制流以及执行栈，但线程在创建它的进程之内运行，分享其父进程的所有数据和系统资源。当应用需要完成并发任务的时候线程是很有用的，但是潜在的问题是任务间必须分享大量的系统状态。
当使用多进程或多线程时，操作系统负责调度。这是通过给每个进程(或线程)一个很小的时间片并且在所有活动任务之间快速循环切换来实现的，这个过程将CPU时间分割为小片段分给各个任务。例如，如果你的系统中有10个活跃的进程正在执行，操作系统将会适当的将十分之一的CPU时间分配给每个进程并且循环地在十个进程之间切换。当系统不止有一个CPU核时，操作系统能够将进程调度到不同的CPU核上，保持系统负载平均以实现并行执行。
利用并发执行机制写的程序需要考虑一些复杂的问题。复杂性的主要来源是关于同步和共享数据的问题。通常情况下，多个任务同时试图更新同一个数据结构会造成脏数据和程序状态不一致的问题(正式的说法是资源竞争的问题)。为了解决这个问题，需要使用互斥锁或是其他相似的同步原语来标识并保护程序中的关键部分。举个例子，如果多个不同的线程正在试图同时向同一个文件写入数据，那么你需要一个互斥锁使这些写操作依次执行，当一个线程在写入时，其他线程必须等待直到当前线程释放这个资源。
Python中的并发编程
Python长久以来一直支持不同方式的并发编程，包括线程、子进程以及其他利用生成器(generatorfunction)的并发实现。
Python在大部分系统上同时支持消息传递和基于线程的并发编程机制。虽然大部分程序员对线程接口更为熟悉，但是Python的线程机制却有着诸多的限制。Python使用了内部全局解释器锁(GIL)来保证线程安全，GIL同时只允许一个线程执行。这使得Python程序就算在多核系统上也只能在单个处理器上运行。Python界关于GIL的争论尽管很多，但在可预见的未来却没有将其移除的可能。
Python提供了一些很精巧的工具用于管理基于线程和进程的并发操作。即使是简单地程序也能够使用这些工具使得任务并发进行从而加快运行速度。subprocess模块为子进程的创建和通信提供了API。这特别适合运行与文本相关的程序，因为这些API支持通过新进程的标准输入输出通道传送数据。signal模块将UNIX系统的信号量机制暴露给用户，用以在进程之间传递事件信息。信号是异步处理的，通常有信号到来时会中断程序当前的工作。信号机制能够实现粗粒度的消息传递系统，但是有其他更可靠的进程内通讯技术能够传递更复杂的消息。threading模块为并发操作提供了一系列高级的，面向对象的API。Thread对象们在一个进程内并发地运行，分享内存资源。使用线程能够更好地扩展I/O密集型的任务。multiprocessing模块同threading模块类似，不过它提供了对于进程的操作。每个进程类是真实的操作系统进程，并且没有共享内存资源，但multiprocessing模块提供了进程间共享数据以及传递消息的机制。通常情况下，将基于线程的程序改为基于进程的很简单，只需要修改一些import声明即可。
Threading模块示例
以threading模块为例，思考这样一个简单的问题：如何使用分段并行的方式完成一个大数的累加。
importthreading classSummingThread(threading.Thread): def__init__(self,low,high): super(SummingThread,self).__init__() self.low=low self.high=high self.total=0 defrun(self): foriinrange(self.low,self.high): self.total+=i thread1=SummingThread(0,500000) thread2=SummingThread(500000,1000000) thread1.start()#Thisactuallycausesthethreadtorun thread2.start() thread1.join()#Thiswaitsuntilthethreadhascompleted thread2.join() #Atthispoint,boththreadshavecompleted result=thread1.total+thread2.total print(result)
自定义Threading类库
我写了一个易于使用threads的小型Python类库，包含了一些有用的类和函数。
关键参数：
*do_threaded_work–该函数将一系列给定的任务分配给对应的处理函数(分配顺序不确定)
*ThreadedWorker–该类创建一个线程，它将从一个同步的工作队列中拉取工作任务并将处理结果写入同步结果队列
*start_logging_with_thread_info–将线程id写入所有日志消息。(依赖日志环境)
*stop_logging_with_thread_info–用于将线程id从所有的日志消息中移除。(依赖日志环境)
importthreading
importlogging
defdo_threaded_work(work_items,work_func,num_threads=None,per_sync_timeout=1,preserve_result_ordering=True):
"""Executeswork_funconeachwork_item.Note:Executionorderisnotpreserved,butoutputorderingis(optionally).
Parameters:
-num_threadsDefault:len(work_items)---Numberofthreadstouseprocessitemsinwork_items.
-per_sync_timeoutDefault:1---Eachsynchronizedoperationcanoptionallytimeout.
-preserve_result_orderingDefault:True---Reordersresult_itemtomatchoriginalwork_itemsordering.
Return:
---listofresultsfromapplyingwork_functoeachwork_item.Orderisoptionallypreserved.
Example:
defprocess_url(url):
#TODO:Dosomeworkwiththeurl
returnurl
urls_to_process=["http://url1.com","http://url2.com","http://site1.com","http://site2.com"]
#processurlsinparallel
result_items=do_threaded_work(urls_to_process,process_url)
#print(results)
print(repr(result_items))
"""
globalwrapped_work_func
ifnotnum_threads:
num_threads=len(work_items)
work_queue=Queue.Queue()
result_queue=Queue.Queue()
index=0
forwork_iteminwork_items:
ifpreserve_result_ordering:
work_queue.put((index,work_item))
else:
work_queue.put(work_item)
index+=1
ifpreserve_result_ordering:
wrapped_work_func=lambdawork_item:(work_item[0],work_func(work_item[1]))
start_logging_with_thread_info()
#spawnapoolofthreads,andpassthemqueueinstance
for_inrange(num_threads):
ifpreserve_result_ordering:
t=ThreadedWorker(work_queue,result_queue,work_func=wrapped_work_func,queue_timeout=per_sync_timeout)
else:
t=ThreadedWorker(work_queue,result_queue,work_func=work_func,queue_timeout=per_sync_timeout)
t.setDaemon(True)
t.start()
work_queue.join()
stop_logging_with_thread_info()
logging.info('work_queuejoined')
result_items=[]
whilenotresult_queue.empty():
result=result_queue.get(timeout=per_sync_timeout)
logging.info('foundresult[:500]:'+repr(result)[:500])
ifresult:
result_items.append(result)
ifpreserve_result_ordering:
result_items=[work_itemforindex,work_iteminresult_items]
returnresult_items
classThreadedWorker(threading.Thread):
"""GenericThreadedWorker
Inputtowork_func:itemfromwork_queue
Exampleusage:
importQueue
urls_to_process=["http://url1.com","http://url2.com","http://site1.com","http://site2.com"]
work_queue=Queue.Queue()
result_queue=Queue.Queue()
defprocess_url(url):
#TODO:Dosomeworkwiththeurl
returnurl
defmain():
#spawnapoolofthreads,andpassthemqueueinstance
foriinrange(3):
t=ThreadedWorker(work_queue,result_queue,work_func=process_url)
t.setDaemon(True)
t.start()
#populatequeuewithdata
forurlinurls_to_process:
work_queue.put(url)
#waitonthequeueuntileverythinghasbeenprocessed
work_queue.join()
#printresults
printrepr(result_queue)
main()
"""
def__init__(self,work_queue,result_queue,work_func,stop_when_work_queue_empty=True,queue_timeout=1):
threading.Thread.__init__(self)
self.work_queue=work_queue
self.result_queue=result_queue
self.work_func=work_func
self.stop_when_work_queue_empty=stop_when_work_queue_empty
self.queue_timeout=queue_timeout
defshould_continue_running(self):
ifself.stop_when_work_queue_empty:
returnnotself.work_queue.empty()
else:
returnTrue
defrun(self):
whileself.should_continue_running():
try:
#grabsitemfromwork_queue
work_item=self.work_queue.get(timeout=self.queue_timeout)
#worksonitem
work_result=self.work_func(work_item)
#placework_resultintoresult_queue
self.result_queue.put(work_result,timeout=self.queue_timeout)
exceptQueue.Empty:
logging.warning('ThreadedWorkerQueuewasemptyorQueue.get()timedout')
exceptQueue.Full:
logging.warning('ThreadedWorkerQueuewasfullorQueue.put()timedout')
except:
logging.exception('ErrorinThreadedWorker')
finally:
#signalstowork_queuethatitemisdone
self.work_queue.task_done()
defstart_logging_with_thread_info():
try:
formatter=logging.Formatter('[thread%(thread)-3s]%(message)s')
logging.getLogger().handlers[0].setFormatter(formatter)
except:
logging.exception('Failedtostartloggingwiththreadinfo')
defstop_logging_with_thread_info():
try:
formatter=logging.Formatter('%(message)s')
logging.getLogger().handlers[0].setFormatter(formatter)
except:
logging.exception('Failedtostoploggingwiththreadinfo')
使用示例
fromtestimportThreadedWorker fromqueueimportQueue urls_to_process=["http://facebook.com","http://pypix.com"] work_queue=Queue() result_queue=Queue() defprocess_url(url): #TODO:Dosomeworkwiththeurl returnurl defmain(): #spawnapoolofthreads,andpassthemqueueinstance foriinrange(5): t=ThreadedWorker(work_queue,result_queue,work_func=process_url) t.setDaemon(True) t.start() #populatequeuewithdata forurlinurls_to_process: work_queue.put(url) #waitonthequeueuntileverythinghasbeenprocessed work_queue.join() #printresults print(repr(result_queue)) main()