本文共 6981 字，大约阅读时间需要 23 分钟。
多进程
创建子进程
import os
print('Process(%s) start...' % os.getpid())
pid = os.fork()
if pid == 0:
print("I am child process (%s) and my parent is %s.' %(os.getpid(), os.getppid()))
else:
print('I (%s) just created a child process(%s).' %(os.getpid(), pid))
python创建子进程是封装了系统的fork调用。
python中创建跨平台的多进程应用，使用multiprocessing模块。
from multiprocessing import Process
import os
def run_proc(name):
print("Run child process %s (%s)...' % (name, os.getpid()))
if __name__='__main__':
print("Parent process %s.' % os.getpid())
p = Process(target=run_proc, args = ('test', ))
print('child process will start.')
p.start()
p.join()
print('child process end.')
创建子进程，只需要传入一个执行函数和函数的参数，创建一个process实例用start()方法启动，比fork()简单。join()可以等待子进程结束后再继续往下运行，用于进程间的同步。
Pool -- 进程池
from multiprocessing import Pool
import os, time, random
def long_time_task(name):
print("Run task %s (%s)...' % (name, os.getpid()))
start = time.time()
time.sleep(random.random() * 3)
end = time.time()
print("Task %s run %0.2f seconds.' % (name, (end - start)))
if __name__ == '__main__':
print("Parent process %s.' % os.getpid())
p = Pool(4)
for i in range(5):
p.apply_async(long_time_task, args=(i,))
print("Waiting for all subprocesses done...')
p.close()
p.join()
print('All subprocesses done.')
Pool对象调用join()会等待所有子进程执行完毕，调用join()之前必须先调用close()，调用close()之后就不能继续添加新的process了。
subprocess -- 启动一个子进程，控制其输入和输出。
子进程需要输入则使用communicate()方法
进程间通信 -- Queue Pipes
在父进程中创建两个子进程，一个向queue写，一个从queuq读。
from multiprocessing import Process, Queue
import os, time, random
def write(q):
print("Process to write: %s' % os.getpid())
for value in ['A', 'B', 'C']:
print('put %s to queue...' % value)
q.put(value)
time.sleep(random.random())
def read(q):
print('Process to read: %s' % os.getpid())
while True:
value = q.get(True)
print('Get %s from queue.' % value)
if __name__ == '__main__':
q = Queue()
pw = Process(target=write, args=(q,))
pr = Process(target=read, args=(q,))
pw.start()
pr.start()
pw.join()
pr.terminate()
多线程
多线程模块 _thread和threading，threading是高级模块，对_thread进行了封装。
启动一个线程就是把一个函数传入并创建Thread实例，然后调用start()开始执行：
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 | import time, threading # 新线程执行的代码:def loop(): print ( 'thread %s is running...' % threading.current_thread().name) n = 0 while n < 5 : n = n + 1 print ( 'thread %s >>> %s' % (threading.current_thread().name, n)) time.sleep( 1 ) print ( 'thread %s ended.' % threading.current_thread().name) print ( 'thread %s is running...' % threading.current_thread().name) t = threading.Thread(target = loop, name = 'LoopThread' ) t.start() t.join() print ( 'thread %s ended.' % threading.current_thread().name) |
Lock
线程锁 --
线程是共享进程数据的，为了保证数据的安全性，必须要使用线程锁。
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 | import time, threading balance = 0 lock = threading.Lock() def change_it(n): global balance balance = balance + n balance = balance - n def run_thread(n): for i in range ( 100000 ): # 先要获取锁 lock.acquire() try : change_it(n) # 一定要释放锁 finally : lock.release() t1 = threading.Thread(target = run_thread,args = ( 5 ,)) t2 = threading.Thread(target = run_thread,args = ( 8 ,)) t1.start() t2.start() t1.join() t2.join() print (balance) |
多个线程同时执行lock.acquire()时，只有一个线程能成功地获取锁，然后继续执行代码，其他线程就继续等待直到获得锁为止。
死锁 -- 由于可以存在多个锁，不同的线程持有不同的锁，并试图获取对方持有的锁时，可能会造成死锁
GIL锁 -- Global Interpreter Lock 任何Python线程执行前，必须先获得GIL锁，然后，每执行100条字节码，解释器就自动释放GIL锁，让别的线程有机会执行。这个GIL全局锁实际上把所有线程的执行代码都给上了锁，所以，多线程在Python中只能交替执行。
解决GIL锁的问题 -- 创建多进程。
问题的抛出：线程的全局变量和局部变量。局部变量只有属于自己的线程才能看到，全局变量的使用必须要加锁。局部变量的传递很麻烦。
解决上述问题使用 -- ThreadLocal
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 | import threading local_school = threading.local() def process_student(): std = local_school.student print ( "hello, %s (in %s)" % (std, threading.current_thread().name)) def process_thread(name): local_school.student = name process_student() t1 = threading.Thread(target = process_thread, args = ( "alice" ,), name = 'Thread-A' ) t2 = threading.Thread(target = process_thread,args = ( "bob" ,), name = 'Thread-B' ) t1.start() t2.start() t1.join() t2.join() |
全局变量local_school就是一个ThreadLocal对象，每个Thread对它都可以读写student属性，但互不影响。你可以把local_school看成全局变量，但每个属性如local_school.student都是线程的局部变量，可以任意读写而互不干扰，也不用管理锁的问题，ThreadLocal内部会处理。
ThreadLocal最常用的地方就是为每个线程绑定一个数据库连接，HTTP请求，用户身份信息等，这样一个线程的所有调用到的处理函数都可以非常方便地访问这些资源。
异步IO
充分利用操作系统提供的异步IO支持，就可以用单进程单线程模型来执行多任务，这种全新的模型称为事件驱动模型
Python中，单进程的异步编程模型称为协程，有了协程的支持，就可以基于事件驱动编写高效的多任务程序。
线程和进程优选进程，进程更稳定进程可以在各个不同的机器上，线程只能在同一个机器上。
multiprocessing模块的managers子模块支持把多进程分布到多台机器上。
一个服务进程可以作为调度者，将任务分布到其他多个进程中，依靠网络通信。
通过消息队列Queue基于managers模块，让其他机器的进程访问Queue
服务进程负责启动Queue，把Queue注册到网络上，然后往Queue里面写入任务：
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 | # task_master.py import random, time, queue from multiprocessing.manager import BaseManager # 发送任务的队列: tast_queue = queue.Queue() # 接收结果的队列 result_queue = queue.Queue() # 从BaseManager继承的QueueManager: class QueueManager(BaseManager): pass # 把两个Queue都注册到网络上, callable参数关联了Queue对象: QueueManager.register( 'get_task_queue' , callable = lambda :task_queue) QueueManager.register( 'get_result_queue' , callable = lambda : result_queue) # 绑定端口5000, 设置验证码'abc': manager = QueueManager(address = (' ', 5000), authkey=b' abc') # 启动Queue: manager.start() # 获得通过网络访问的Queue对象: task = manager.get_task_queue() result = manager.get_result_queue() # 放几个任务进去: for i in range ( 10 ): n = random.randint( 0 , 10000 ) print ( 'put task %d...' % n) task.put(n) # 从result队列读取结果: print ( 'Try get results...' ) for i in range ( 10 ): r = result.get(timeout = 10 ) print ( 'Result: %s' % r) # 关闭: manager.shutdown() print ( 'master exit.' ) |
分布式多进程环境下，添加任务到Queue不可以直接对原始的task_queue进行操作，必须通过manager.get_task_queue()获得的Queue接口添加。
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 | # task_worker.py import time, sys, queue from multiprocessing.managers import BaseManager # 创建类似的QueueManager: class QueueManager(BaseManager): pass # 由于这个QueueManager只从网络上获取Queue，所以注册时只提供名字: QueueManager.register( 'get_task_queue' ) QueueManager.register( 'get_result_queue' ) # 连接到服务器，也就是运行task_master.py的机器: server_addr = '127.0.0.1' print ( 'Connect to server %s...' % server_addr) # 端口和验证码注意保持与task_master.py设置的完全一致: m = QueueManager(address = (server_addr, 5000 ), authkey = b 'abc' ) # 从网络连接: m.connect() # 获取Queue的对象: task = m.get_task_queue() result = m.get_result_queue() # 从task队列取任务,并把结果写入result队列: for i in range ( 10 ): try : n = task.get(timeout = 1 ) print ( 'run task %d * %d...' % (n, n)) r = '%d * %d = %d' % (n, n, n * n) time.sleep( 1 ) result.put(r) except Queue.Empty: print ( 'task queue is empty.' ) # 处理结束: print ( 'worker exit.' ) |
Queue能通过网络访问，就是通过QueueManager实现的。由于QueueManager管理的不止一个Queue，所以，要给每个Queue的网络调用接口起个名字，比如get_task_queue。
authkey有什么用？这是为了保证两台机器正常通信，不被其他机器恶意干扰。如果task_worker.py的authkey和task_master.py的authkey不一致，肯定连接不上。