python線程、進程和協程
- 高洛峰原創
- 2017-02-28 16:34:481240瀏覽
Python被人詬病最多的大概就是效能差，在這裡講一下 Python 的多進程，多執行緒與協程。先聲明這不是教程，看完這篇文章，大概能夠對 Python 的多進程與多線程有一定的了解。
引言
解釋器環境：python3.5.1
我們都知道python網路程式設計的兩大必學模組socket和socketserver，其中的socketserver是一個支援IO多工和多執行緒、多進程的模組。一般我們在socketserver服務端程式碼中都會寫這麼一句：
server = socketserver.ThreadingTCPServer(settings.IP_PORT, MyServer)
ThreadingTCPServer這個類別是一個支援多執行緒和TCP協定的socketserver，它的繼承關係是這樣的：
class ThreadingTCPServer(ThreadingMixIn, TCPServer): pass
右邊的TCPServer其實是它主要的功能父類，而左邊的ThreadingMixIn則是實作了多執行緒的類，它本身本身則沒有任何程式碼。
MixIn在python的類別命名中，很常見，一般被稱為“混入”，戲稱“亂入”，通常為了某種重要功能被子類別繼承。
class ThreadingMixIn:
daemon_threads = False
def process_request_thread(self, request, client_address):
try:
self.finish_request(request, client_address)
self.shutdown_request(request)
except:
self.handle_error(request, client_address)
self.shutdown_request(request)
def process_request(self, request, client_address):
t = threading.Thread(target = self.process_request_thread,
args = (request, client_address))
t.daemon = self.daemon_threads
t.start()
在ThreadingMixIn類別中，其實就定義了一個屬性，兩個方法。在process_request方法中實際呼叫的正是python內建的多執行緒模組threading。這個模組是python中所有多執行緒的基礎，socketserver本質上也是利用了這個模組。
一、執行緒
#複製程式碼 程式碼如下：
#線程，有時被稱為輕量級進程(Lightweight Process，LWP），是程式執行流的最小單元。一個標準的執行緒由執行緒ID，目前指令指標(PC），暫存器集合和堆疊組成。另外，執行緒是進程中的一個實體，是被系統獨立調度和分派的基本單位，執行緒本身不獨立擁有系統資源，但它可與同屬一個進程的其它執行緒共享該進程所擁有的全部資源。一個線程可以建立和撤銷另一個線程，同一進程中的多個線程之間可以並發執行。由於執行緒之間的相互制約，致使執行緒在運行中呈現出間斷性。線程也有就緒、阻斷和運行三種基本狀態。就緒狀態是指執行緒具備運作的所有條件，邏輯上可以運行，在等待處理機；運作狀態是指執行緒佔有處理機正在運作；阻塞狀態是指執行緒在等待一個事件（如某個訊號量），邏輯上不可執行。每一個應用程式都至少有一個行程和一個執行緒。執行緒是程式中一個單一的順序控制流程。在單一程式中同時執行多個執行緒完成不同的被分割成一塊一塊的工作，稱為多執行緒。
以上那一段，可以不用看！舉個例子，廠商要生產某個產品，在它的生產基地建造了許多廠房，每個廠房內有多條流水生產線。所有廠房配合將整個產品生產出來，某廠房內的所有流水線將這個廠房負責的產品部分生產出來。每個廠房擁有自己的材料庫，廠房內的生產線共享這些材料。而每一個廠商要實現生產必須擁有至少一個廠房一條生產線。那麼這個廠家就是某個應用程式；每個廠房就是一個行程；每條生產線都是一個執行緒。
1.1 普通的多執行緒
在python中，threading模組提供執行緒的功能。透過它，我們可以輕易的在進程中創建多個執行緒。以下是個例子：
import threading import time def show(arg): time.sleep(1) print('thread'+str(arg)) for i in range(10): t = threading.Thread(target=show, args=(i,)) t.start() print('main thread stop')
上述程式碼建立了10個「前台」線程，然後控制器就交給了CPU，CPU根據指定演算法進行調度，分片執行指令。
以下是Thread類別的主要方法：
start 執行緒準備就緒，等待CPU調度
setName 為執行緒設定名稱
getName 取得執行緒名稱
setDaemon設定為後台線程或前台線程（預設）
如果是後台線程，主線程執行過程中，後台線程也在進行，主線程執行完畢後，後台線程不論成功與否，均停止。如果是前台線程，主線程執行過程中，前台線程也在進行，主線程執行完畢後，等待前台線程也執行完成後，程式停止。
join 逐一執行每個線程，執行完畢後繼續往下執行，該方法使得多線程變得無意義。
run 執行緒被cpu調度後自動執行執行緒物件的run方法
#1.2 自訂執行緒類別
對於threading模組中的Thread類，本質上是執行了它的run方法。因此可以自訂線程類，讓它繼承Thread類，然後重寫run方法。
import threading class MyThreading(threading.Thread): def __init__(self,func,arg): super(MyThreading,self).__init__() self.func = func self.arg = arg def run(self): self.func(self.arg) def f1(args): print(args) obj = MyThreading(f1, 123) obj.start()
1.3 執行緒鎖定
CPU执行任务时，在线程之间是进行随机调度的，并且每个线程可能只执行n条代码后就转而执行另外一条线程。由于在一个进程中的多个线程之间是共享资源和数据的，这就容易造成资源抢夺或脏数据，于是就有了锁的概念，限制某一时刻只有一个线程能访问某个指定的数据。
1.3.1 未使用锁
#!/usr/bin/env python # -*- coding:utf-8 -*- import threading import time NUM = 0 def show(): global NUM NUM += 1 name = t.getName() time.sleep(1) # 注意，这行语句的位置很重要，必须在NUM被修改后，否则观察不到脏数据的现象。 print(name, "执行完毕后，NUM的值为： ", NUM) for i in range(10): t = threading.Thread(target=show) t.start() print('main thread stop')
上述代码运行后，结果如下：
main thread stop Thread-1 执行完毕后，NUM的值为： 10 Thread-2 执行完毕后，NUM的值为： 10 Thread-4 执行完毕后，NUM的值为： 10 Thread-9 执行完毕后，NUM的值为： 10 Thread-3 执行完毕后，NUM的值为： 10 Thread-6 执行完毕后，NUM的值为： 10 Thread-8 执行完毕后，NUM的值为： 10 Thread-7 执行完毕后，NUM的值为： 10 Thread-5 执行完毕后，NUM的值为： 10 Thread-10 执行完毕后，NUM的值为： 10
由此可见，由于线程同时访问一个数据，产生了错误的结果。为了解决这个问题，python在threading模块中定义了几种线程锁类，分别是：
Lock 普通锁（不可嵌套）
RLock 普通锁（可嵌套）常用
Semaphore 信号量
event 事件
condition 条件
1.3.2 普通锁Lock和RLock
类名：Lock或RLock
普通锁，也叫互斥锁，是独占的，同一时刻只有一个线程被放行。
import time import threading NUM = 10 def func(lock): global NUM lock.acquire() # 让锁开始起作用 NUM -= 1 time.sleep(1) print(NUM) lock.release() # 释放锁 lock = threading.Lock() # 实例化一个锁对象 for i in range(10): t = threading.Thread(target=func, args=(lock,)) # 记得把锁当作参数传递给func参数 t.start()
以上是threading模块的Lock类，它不支持嵌套锁。RLcok类的用法和Lock一模一样，但它支持嵌套，因此我们一般直接使用RLcok类。
1.3.3 信号量(Semaphore)
类名：BoundedSemaphore
这种锁允许一定数量的线程同时更改数据，它不是互斥锁。比如地铁安检，排队人很多，工作人员只允许一定数量的人进入安检区，其它的人继续排队。
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import time
import threading
def run(n):
semaphore.acquire()
print("run the thread: %s" % n)
time.sleep(1)
semaphore.release()
num = 0
semaphore = threading.BoundedSemaphore(5) # 最多允许5个线程同时运行
for i in range(20):
t = threading.Thread(target=run, args=(i,))
t.start()
1.3.4 事件(Event)
类名：Event
事件主要提供了三个方法 set、wait、clear。
事件机制：全局定义了一个“Flag”，如果“Flag”的值为False，那么当程序执行wait方法时就会阻塞，如果“Flag”值为True，那么wait方法时便不再阻塞。这种锁，类似交通红绿灯（默认是红灯），它属于在红灯的时候一次性阻挡所有线程，在绿灯的时候，一次性放行所有的排队中的线程。
clear：将“Flag”设置为False
set：将“Flag”设置为True
import threading
def func(e,i):
print(i)
e.wait() # 检测当前event是什么状态，如果是红灯，则阻塞，如果是绿灯则继续往下执行。默认是红灯。
print(i+100)
event = threading.Event()
for i in range(10):
t = threading.Thread(target=func, args=(event, i))
t.start()
event.clear() # 主动将状态设置为红灯
inp = input(">>>")
if inp == "1":
event.set() # 主动将状态设置为绿灯
1.3.5 条件(condition)
类名：Condition
该机制会使得线程等待，只有满足某条件时，才释放n个线程。
import threading
def condition():
ret = False
r = input(">>>")
if r == "yes":
ret = True
return ret
def func(conn, i):
print(i)
conn.acquire()
conn.wait_for(condition) # 这个方法接受一个函数的返回值
print(i+100)
conn.release()
c = threading.Condition()
for i in range(10):
t = threading.Thread(target=func, args=(c, i,))
t.start()
上面的例子，每输入一次“yes”放行了一个线程。下面这个，可以选择一次放行几个线程。
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import threading
def run(n):
con.acquire()
con.wait()
print("run the thread: %s" %n)
con.release()
if __name__ == '__main__':
con = threading.Condition()
for i in range(10):
t = threading.Thread(target=run, args=(i,))
t.start()
while True:
inp = input('>>>')
if inp == "q":
break
# 下面这三行是固定语法
con.acquire()
con.notify(int(inp)) # 这个方法接收一个整数，表示让多少个线程通过
con.release()
1.3 全局解释器锁（GIL）
既然介绍了多线程和线程锁，那就不得不提及python的GIL，也就是全局解释器锁。在编程语言的世界，python因为GIL的问题广受诟病，因为它在解释器的层面限制了程序在同一时间只有一个线程被CPU实际执行，而不管你的程序里实际开了多少条线程。所以我们经常能发现，python中的多线程编程有时候效率还不如单线程，就是因为这个原因。那么，对于这个GIL，一些普遍的问题如下：
每种编程语言都有GIL吗？
以python官方Cpython解释器为代表....其他语言好像未见。
为什么要有GIL？
作为解释型语言，Python的解释器必须做到既安全又高效。我们都知道多线程编程会遇到的问题。解释器要留意的是避免在不同的线程操作内部共享的数据。同时它还要保证在管理用户线程时总是有最大化的计算资源。那么，不同线程同时访问时，数据的保护机制是怎样的呢？答案是解释器全局锁GIL。GIL对诸如当前线程状态和为垃圾回收而用的堆分配对象这样的东西的访问提供着保护。
为什么不能去掉GIL？
首先，在早期的python解释器依赖较多的全局状态，传承下来，使得想要移除当今的GIL变得更加困难。其次，对于程序员而言，仅仅是想要理解它的实现就需要对操作系统设计、多线程编程、C语言、解释器设计和CPython解释器的实现有着非常彻底的理解。
在1999年，针对Python1.5，一个“freethreading”补丁已经尝试移除GIL，用细粒度的锁来代替。然而，GIL的移除给单线程程序的执行速度带来了一定的负面影响。当用单线程执行时，速度大约降低了40%。虽然使用两个线程时在速度上得到了提高，但这个提高并没有随着核数的增加而线性增长。因此这个补丁没有被采纳。
另外，在python的不同解释器实现中，如PyPy就移除了GIL，其执行速度更快（不单单是去除GIL的原因）。然而，我们通常使用的CPython占有着统治地位的使用量，所以，你懂的。
在Python 3.2中实现了一个新的GIL，并且带着一些积极的结果。这是自1992年以来，GIL的一次最主要改变。旧的GIL通过对Python指令进行计数来确定何时放弃GIL。在新的GIL实现中，用一个固定的超时时间来指示当前的线程以放弃这个锁。在当前线程保持这个锁，且当第二个线程请求这个锁的时候，当前线程就会在5ms后被强制释放掉这个锁（这就是说，当前线程每5ms就要检查其是否需要释放这个锁）。当任务是可行的时候，这会使得线程间的切换更加可预测。
GIL对我们有什么影响？
最大的影响是我们不能随意使用多线程。要区分任务场景。
在单核cpu情况下对性能的影响可以忽略不计，多线程多进程都差不多。在多核CPU时，多线程效率较低。GIL对单进程和多进程没有影响。
在实际使用中有什么好的建议？
建议在IO密集型任务中使用多线程，在计算密集型任务中使用多进程。深入研究python的协程机制，你会有惊喜的。
更多的详细介绍和说明请参考下面的文献：
原文：Python's Hardest Problem
译文：Python 最难的问题
1.4 定时器（Timer）
定时器，指定n秒后执行某操作。很简单但很使用的东西。
from threading import Timer
def hello():
print("hello, world")
t = Timer(1, hello) # 表示1秒后执行hello函数
t.start()
1.5 队列
通常而言，队列是一种先进先出的数据结构，与之对应的是堆栈这种后进先出的结构。但是在python中，它内置了一个queue模块，它不但提供普通的队列，还提供一些特殊的队列。具体如下：
queue.Queue ：先进先出队列
queue.LifoQueue ：后进先出队列
queue.PriorityQueue ：优先级队列
queue.deque ：双向队列
1.5.1 Queue：先进先出队列
这是最常用也是最普遍的队列，先看一个例子。
import queue q = queue.Queue(5) q.put(11) q.put(22) q.put(33) print(q.get()) print(q.get()) print(q.get())
Queue类的参数和方法：
maxsize 队列的最大元素个数，也就是queue.Queue(5)中的5。当队列内的元素达到这个值时，后来的元素默认会阻塞，等待队列腾出位置。
def __init__(self, maxsize=0): self.maxsize = maxsize self._init(maxsize)
qsize() 获取当前队列中元素的个数，也就是队列的大小empty() 判断当前队列是否为空，返回True或者Falsefull() 判断当前队列是否已满，返回True或者False
put(self, block=True, timeout=None)
往队列里放一个元素，默认是阻塞和无时间限制的。如果，block设置为False，则不阻塞，这时，如果队列是满的，放不进去，就会弹出异常。如果timeout设置为n秒，则会等待这个秒数后才put，如果put不进去则弹出异常。
get(self, block=True, timeout=None)
从队列里获取一个元素。参数和put是一样的意思。
join() 阻塞进程，直到所有任务完成，需要配合另一个方法task_done。
def join(self): with self.all_tasks_done: while self.unfinished_tasks: self.all_tasks_done.wait()
task_done() 表示某个任务完成。每一条get语句后需要一条task_done。
import queue q = queue.Queue(5) q.put(11) q.put(22) print(q.get()) q.task_done() print(q.get()) q.task_done() q.join()
1.5.2 LifoQueue：后进先出队列
类似于“堆栈”，后进先出。也较常用。
import queue q = queue.LifoQueue() q.put(123) q.put(456) print(q.get())
上述代码运行结果是：456
1.5.3 PriorityQueue：优先级队列
带有权重的队列，每个元素都是一个元组，前面的数字表示它的优先级，数字越小优先级越高，同样的优先级先进先出
q = queue.PriorityQueue() q.put((1,"alex1")) q.put((1,"alex2")) q.put((1,"alex3")) q.put((3,"alex3")) print(q.get())
1.5.4 deque：双向队列
Queue和LifoQueue的“综合体”，双向进出。方法较多，使用复杂，慎用！
q = queue.deque() q.append(123) q.append(333) q.appendleft(456) q.pop() q.popleft()
1.6 生产者消费者模型
利用多线程和队列可以搭建一个生产者消费者模型，用于处理大并发的服务。
在并发编程中使用生产者和消费者模式能够解决绝大多数并发问题。该模式通过平衡生产线程和消费线程的工作能力来提高程序的整体处理数据的速度。
为什么要使用生产者和消费者模式
在线程世界里，生产者就是生产数据的线程，消费者就是消费数据的线程。在多线程开发当中，如果生产者处理速度很快，而消费者处理速度很慢，那么生产者就必须等待消费者处理完，才能继续生产数据。同样的道理，如果消费者的处理能力大于生产者，那么消费者就必须等待生产者。为了解决这个问题于是引入了生产者和消费者模式。
什么是生产者消费者模式
生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。
这个阻塞队列就是用来给生产者和消费者解耦的。纵观大多数设计模式，都会找一个第三者出来进行解耦，如工厂模式的第三者是工厂类，模板模式的第三者是模板类。在学习一些设计模式的过程中，如果先找到这个模式的第三者，能帮助我们快速熟悉一个设计模式。
以上摘自方腾飞的《聊聊并发——生产者消费者模式》
下面是一个简单的厨师做包子，顾客吃包子的例子。
#!/usr/bin/env python
# -*- coding:utf-8 -*-
# Author:Liu Jiang
import time
import queue
import threading
q = queue.Queue(10)
def productor(i):
while True:
q.put("厨师 %s 做的包子！"%i)
time.sleep(2)
def consumer(k):
while True:
print("顾客 %s 吃了一个 %s"%(k,q.get()))
time.sleep(1)
for i in range(3):
t = threading.Thread(target=productor,args=(i,))
t.start()
for k in range(10):
v = threading.Thread(target=consumer,args=(k,))
v.start()
1.7 线程池
在使用多线程处理任务时也不是线程越多越好，由于在切换线程的时候，需要切换上下文环境，依然会造成cpu的大量开销。为解决这个问题，线程池的概念被提出来了。预先创建好一个较为优化的数量的线程，让过来的任务立刻能够使用，就形成了线程池。在python中，没有内置的较好的线程池模块，需要自己实现或使用第三方模块。下面是一个简单的线程池：
#!/usr/bin/env python # -*- coding:utf-8 -*- # Author:Liu Jiang import queue import time import threading class MyThreadPool: def __init__(self, maxsize=5): self.maxsize = maxsize self._q = queue.Queue(maxsize) for i in range(maxsize): self._q.put(threading.Thread) def get_thread(self): return self._q.get() def add_thread(self): self._q.put(threading.Thread) def task(i, pool): print(i) time.sleep(1) pool.add_thread() pool = MyThreadPool(5) for i in range(100): t = pool.get_thread() obj = t(target=task, args=(i,pool)) obj.start()
上面的例子是把线程类当做元素添加到队列内。实现方法比较糙，每个线程使用后就被抛弃，一开始就将线程开到满，因此性能较差。下面是一个相对好一点的例子：
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import queue
import threading
import contextlib
import time
StopEvent = object() # 创建空对象
class ThreadPool(object):
def __init__(self, max_num, max_task_num = None):
if max_task_num:
self.q = queue.Queue(max_task_num)
else:
self.q = queue.Queue()
self.max_num = max_num
self.cancel = False
self.terminal = False
self.generate_list = []
self.free_list = []
def run(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):
"""
循环去获取任务函数并执行任务函数
"""
current_thread = threading.currentThread
self.generate_list.append(current_thread)
event = self.q.get()
while event != StopEvent:
func, arguments, callback = event
try:
result = func(*arguments)
success = True
except Exception as e:
success = False
result = None
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)
self.q.empty()
@contextlib.contextmanager
def worker_state(self, state_list, worker_thread):
"""
用于记录线程中正在等待的线程数
"""
state_list.append(worker_thread)
try:
yield
finally:
state_list.remove(worker_thread)
# How to use
pool = ThreadPool(5)
def callback(status, result):
# status, execute action status
# result, execute action return value
pass
def action(i):
print(i)
for i in range(30):
ret = pool.run(action, (i,), callback)
time.sleep(5)
print(len(pool.generate_list), len(pool.free_list))
print(len(pool.generate_list), len(pool.free_list))
# pool.close()
# pool.terminate()
二、进程
在python中multiprocess模块提供了Process类，实现进程相关的功能。但是，由于它是基于fork机制的，因此不被windows平台支持。想要在windows中运行，必须使用if __name__ == '__main__:的方式，显然这只能用于调试和学习，不能用于实际环境。
（PS：在这里我必须吐槽一下python的包、模块和类的组织结构。在multiprocess中你既可以import大写的Process，也可以import小写的process，这两者是完全不同的东西。这种情况在python中很多，新手容易傻傻分不清。）
下面是一个简单的多进程例子，你会发现Process的用法和Thread的用法几乎一模一样。
from multiprocessing import Process
def foo(i):
print("This is Process ", i)
if __name__ == '__main__':
for i in range(5):
p = Process(target=foo, args=(i,))
p.start()
2.1 进程的数据共享
每个进程都有自己独立的数据空间，不同进程之间通常是不能共享数据，创建一个进程需要非常大的开销。
from multiprocessing import Process
list_1 = []
def foo(i):
list_1.append(i)
print("This is Process ", i," and list_1 is ", list_1)
if __name__ == '__main__':
for i in range(5):
p = Process(target=foo, args=(i,))
p.start()
print("The end of list_1:", list_1)
运行上面的代码，你会发现列表list_1在各个进程中只有自己的数据，完全无法共享。想要进程之间进行资源共享可以使用queues/Array/Manager这三个multiprocess模块提供的类。
2.1.1 使用Array共享数据
from multiprocessing import Process from multiprocessing import Array def Foo(i,temp): temp[0] += 100 for item in temp: print(i,'----->',item) if __name__ == '__main__': temp = Array('i', [11, 22, 33, 44]) for i in range(2): p = Process(target=Foo, args=(i,temp)) p.start()
对于Array数组类，括号内的“i”表示它内部的元素全部是int类型，而不是指字符i，列表内的元素可以预先指定，也可以指定列表长度。概括的来说就是Array类在实例化的时候就必须指定数组的数据类型和数组的大小，类似temp = Array('i', 5)。对于数据类型有下面的表格对应：
'c': ctypes.c_char, 'u': ctypes.c_wchar,
'b': ctypes.c_byte, 'B': ctypes.c_ubyte,
'h': ctypes.c_short, 'H': ctypes.c_ushort,
'i': ctypes.c_int, 'I': ctypes.c_uint,
'l': ctypes.c_long, 'L': ctypes.c_ulong,
'f': ctypes.c_float, 'd': ctypes.c_double
2.1.2 使用Manager共享数据
from multiprocessing import Process,Manager def Foo(i,dic): dic[i] = 100+i print(dic.values()) if __name__ == '__main__': manage = Manager() dic = manage.dict() for i in range(10): p = Process(target=Foo, args=(i,dic)) p.start() p.join()
Manager比Array要好用一点，因为它可以同时保存多种类型的数据格式。
2.1.3 使用queues的Queue类共享数据
import multiprocessing from multiprocessing import Process from multiprocessing import queues def foo(i,arg): arg.put(i) print('The Process is ', i, "and the queue's size is ", arg.qsize()) if __name__ == "__main__": li = queues.Queue(20, ctx=multiprocessing) for i in range(10): p = Process(target=foo, args=(i,li,)) p.start()
这里就有点类似上面的队列了。从运行结果里，你还能发现数据共享中存在的脏数据问题。另外，比较悲催的是multiprocessing里还有一个Queue，一样能实现这个功能。
2.2 进程锁
为了防止和多线程一样的出现数据抢夺和脏数据的问题，同样需要设置进程锁。与threading类似，在multiprocessing里也有同名的锁类RLock, Lock, Event, Condition, Semaphore，连用法都是一样样的！（这个我喜欢）
from multiprocessing import Process from multiprocessing import queues from multiprocessing import Array from multiprocessing import RLock, Lock, Event, Condition, Semaphore import multiprocessing import time def foo(i,lis,lc): lc.acquire() lis[0] = lis[0] - 1 time.sleep(1) print('say hi',lis[0]) lc.release() if __name__ == "__main__": # li = [] li = Array('i', 1) li[0] = 10 lock = RLock() for i in range(10): p = Process(target=foo,args=(i,li,lock)) p.start()
2.3 进程池
既然有线程池，那必然也有进程池。但是，python给我们内置了一个进程池，不需要像线程池那样需要自定义，你只需要简单的from multiprocessing import Pool。
#!/usr/bin/env python # -*- coding:utf-8 -*- from multiprocessing import Pool import time def f1(args): time.sleep(1) print(args) if __name__ == '__main__': p = Pool(5) for i in range(30): p.apply_async(func=f1, args= (i,)) p.close() # 等子进程执行完毕后关闭线程池 # time.sleep(2) # p.terminate() # 立刻关闭线程池 p.join()
进程池内部维护一个进程序列，当使用时，去进程池中获取一个进程，如果进程池序列中没有可供使用的进程，那么程序就会等待，直到进程池中有可用进程为止。
进程池中有以下几个主要方法：
apply：从进程池里取一个进程并执行
apply_async：apply的异步版本
terminate:立刻关闭线程池
join：主进程等待所有子进程执行完毕，必须在close或terminate之后
close：等待所有进程结束后，才关闭线程池
三、协程
线程和进程的操作是由程序触发系统接口，最后的执行者是系统，它本质上是操作系统提供的功能。而协程的操作则是程序员指定的，在python中通过yield，人为的实现并发处理。
协程存在的意义：对于多线程应用，CPU通过切片的方式来切换线程间的执行，线程切换时需要耗时。协程，则只使用一个线程，分解一个线程成为多个“微线程”，在一个线程中规定某个代码块的执行顺序。
协程的适用场景：当程序中存在大量不需要CPU的操作时（IO）。
在不需要自己“造轮子”的年代，同样有第三方模块为我们提供了高效的协程，这里介绍一下greenlet和gevent。本质上，gevent是对greenlet的高级封装，因此一般用它就行，这是一个相当高效的模块。
在使用它们之前，需要先安装，可以通过源码，也可以通过pip。
3.1 greenlet
from greenlet import greenlet def test1(): print(12) gr2.switch() print(34) gr2.switch() def test2(): print(56) gr1.switch() print(78) gr1 = greenlet(test1) gr2 = greenlet(test2) gr1.switch()
实际上，greenlet就是通过switch方法在不同的任务之间进行切换。
3.2 gevent
from gevent import monkey; monkey.patch_all() import gevent import requests def f(url): print('GET: %s' % url) resp = requests.get(url) data = resp.text print('%d bytes received from %s.' % (len(data), url)) gevent.joinall([ gevent.spawn(f, 'https://www.python.org/'), gevent.spawn(f, 'https://www.yahoo.com/'), gevent.spawn(f, 'https://github.com/'), ])
通过joinall将任务f和它的参数进行统一调度，实现单线程中的协程。代码封装层次很高，实际使用只需要了解它的几个主要方法即可。
更多python线程、进程和协程相关文章请关注PHP中文网！