Python 스레드, 프로세스 및 코루틴
- 高洛峰원래의
- 2017-02-28 16:34:481238검색
Python에 대해 가장 비판받는 점은 아마도 성능이 좋지 않다는 점일 것입니다. 여기서는 Python의 다중 프로세스, 다중 스레드 및 코루틴에 대해 이야기해 보겠습니다. 우선 이 글은 튜토리얼이 아니라는 점을 말씀드리고 싶습니다. 이 글을 읽고 나면 여러분은 아마도 Python의 멀티프로세싱과 멀티스레딩에 대해 어느 정도 이해하게 될 것입니다.
소개
통역사 환경: python3.5.1
우리 모두는 Python 네트워크 프로그래밍을 위해 꼭 배워야 할 두 가지 모듈을 알고 있습니다. 소켓과 소켓서버 중 소켓서버는 IO 멀티플렉싱, 멀티스레딩, 멀티프로세스를 지원하는 모듈이다. 일반적으로 소켓서버 서버 코드에 다음 문장을 작성합니다:
server = socketserver.ThreadingTCPServer(settings.IP_PORT, MyServer)
ThreadingTCPServer이 클래스는 멀티스레딩과 TCP를 지원하는 소켓서버입니다. 프로토콜의 상속 관계는 다음과 같습니다.
class ThreadingTCPServer(ThreadingMixIn, TCPServer): pass
오른쪽의 TCPServer는 실제로 주요 기능 상위 클래스인 반면 왼쪽의 ThreadingMixIn은 구현입니다. 다중 스레드 클래스 자체에는 코드가 없습니다.
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에 내장된 멀티스레딩 모듈 스레딩입니다. 이 모듈은 Python의 모든 멀티스레딩의 기초이며, 소켓서버는 이 모듈을 기본적으로 활용합니다.
1. 스레드
코드 복사 코드는 다음과 같습니다.
경량 프로세스(LWP)라고도 불리는 스레드는 프로그램 실행 흐름의 가장 작은 단위입니다. 표준 스레드는 스레드 ID, 현재 명령어 포인터(PC), 레지스터 세트 및 스택으로 구성됩니다. 또한 스레드는 프로세스 내의 개체로서 시스템에 의해 독립적으로 예약되고 전달되는 기본 단위입니다. 스레드 자체는 시스템 리소스를 독립적으로 소유하지 않지만 프로세스가 소유한 모든 리소스를 다른 스레드와 공유할 수 있습니다. 같은 프로세스에 속합니다. 스레드는 다른 스레드를 생성하고 삭제할 수 있으며, 동일한 프로세스의 여러 스레드가 동시에 실행될 수 있습니다. 스레드 간의 상호 제약으로 인해 스레드는 작업에 불연속성을 나타냅니다. 스레드에는 준비, 차단, 실행이라는 세 가지 기본 상태도 있습니다. 준비 상태는 스레드가 실행할 모든 조건을 갖추고 있고 논리적으로 실행될 수 있으며 프로세서를 기다리고 있음을 의미합니다. 실행 상태는 스레드가 프로세서를 점유하고 실행 중임을 의미합니다. 이벤트(예: 세마포어) 및 논리를 시행할 수 없습니다. 모든 애플리케이션에는 최소한 하나의 프로세스와 하나의 스레드가 있습니다. 스레드는 프로그램의 단일 순차적 제어 흐름입니다. 단일 프로그램에서 여러 스레드를 동시에 실행하여 여러 조각으로 나누어진 여러 작업을 완료하는 것을 멀티스레딩이라고 합니다. 위 문단을 읽을 필요는 없습니다! 예를 들어, 제조업체가 특정 제품을 생산하려면 생산 기지에 많은 공장을 짓고 각 공장에는 여러 생산 라인이 있습니다. 모든 공장이 협력하여 전체 제품을 생산하고, 특정 공장의 모든 조립 라인은 해당 공장이 담당하는 제품의 일부를 생산합니다. 각 공장에는 자체 재료 라이브러리가 있으며, 공장의 생산 라인은 이러한 재료를 공유합니다. 생산을 달성하려면 모든 제조업체는 최소한 하나의 공장 건물과 하나의 생산 라인을 보유해야 합니다. 그러면 이 제조업체는 애플리케이션이고 각 공장은 프로세스입니다.1.1 일반 멀티스레딩
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는 CPU에 따라 예약됩니다. 지정된 알고리즘을 통해 명령을 실행합니다.
setName 스레드 이름 설정
getName 스레드 이름 가져오기
setDaemon 백그라운드 스레드 또는 포그라운드 스레드로 설정(기본값)
백그라운드 스레드인 경우 메인 스레드 실행 중에 메인 스레드 실행이 완료된 후 백그라운드 스레드도 실행됩니다. 성공이나 실패에 관계없이 백그라운드 스레드가 중지됩니다. 포그라운드 스레드인 경우 메인 스레드 실행 중에 포그라운드 스레드도 진행 중입니다. 메인 스레드가 실행을 완료한 후 포그라운드 스레드가 실행을 완료할 때까지 기다린 후 프로그램이 중지됩니다.
join은 각 스레드를 하나씩 실행하고 완료 후에도 실행을 계속합니다. 이 방법은 멀티스레딩을 의미 없게 만듭니다.
run 스레드는 CPU에 의해 예약된 후 자동으로 스레드 개체의 run 메서드를 실행합니다.
1.2 사용자 정의 스레드 클래스
아아앙
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中文网！