python3 多进程同步

进程同步之：Queue和JoinableQueue

代码示例1：Queue

代码示例2：两个子进程间通信

代码示例3：子进程间通信Queue&JoinableQueue

进程同步之：Lock

进程同步之：event

进程同步之：pipe（管道）

进程同步之：Condition

进程同步之：共享变量（数字/字符串/列表/字典/实例对象）

代码示例1：未使用共享变量

代码示例2：共享数字变量

代码示例3：共享变量+锁

代码示例4：共享字符串

代码示例5：共享的字典和列表

代码示例6：共享实例对象

进程同步之：Queue和JoinableQueue

multiprocessing.Queue类似于queue.Queue，一般用来多个进程间交互信息。Queue是进程和线程安全的。它实现了queue.Queue的大部分方法，但task_done()和join()没有实现。
multiprocessing.JoinableQueue是multiprocessing.Queue的子类，增加了task_done()方法和join()方法。
task_done()：一般在调用get()时获得一个task，在task结束后调用task_done()来通知Queue当前task完成。
join()：阻塞直到queue中的所有的task都被处理（即task_done方法被调用）。

代码示例1：Queue

#encoding=utf-8
from multiprocessing import Process, Queue  def offer(queue):  # 入队列if queue.empty():queue.put("Hello World")  else:print(queue.get())if __name__ == '__main__':  # 创建一个队列实例q = Queue() p = Process(target = offer, args = (q,))  p.start()  print(q.get()) # 主进程执行出队列操作q.put("哈哈哈哈哈")# 主进程执行入队列操作m = Process(target = offer, args = (q,)) # 子进程执行入队列操作m.start()m.join()p.join()

代码示例2：两个子进程间通信

#encoding=utf-8
from multiprocessing import Process, Queue
import os, time, random # 写数据进程执行的代码:
def write(q): for value in ['A', 'B', 'C']: print('Put %s to queue...' % value) q.put(value) time.sleep(random.random()) # 读数据进程执行的代码
def read(q):time.sleep(1)while not q.empty():# if not q.empty():print('Get %s from queue.' % q.get(True))time.sleep(1) # 目的是等待写队列完成if __name__=='__main__': # 父进程创建Queue，并传给各个子进程q = Queue()pw = Process(target = write, args = (q,)) pr = Process(target = read, args = (q,)) # 启动子进程pw，写入: pw.start() # 启动子进程pr，读取: pr.start() # 等待pw结束: pw.join() pr.join()
print("Done!")

代码示例3：子进程间通信Queue&JoinableQueue

#encoding=utf-8
import multiprocessing
import time
class Consumer(multiprocessing.Process):# 派生进程def __init__(self, task_queue, result_queue):multiprocessing.Process.__init__(self)self.task_queue = task_queue#任务队列self.result_queue = result_queue#结果队列# 重写原进程的run方法def run(self):proc_name = self.namewhile True:next_task = self.task_queue.get()if next_task is None:#为None,进程退出# Poison pill means shutdownprint(('%s: Exiting' % proc_name))self.task_queue.task_done()breakprint(('%s: %s' % (proc_name, next_task)))answer = next_task() # __call__()self.task_queue.task_done()#执行完get之后调用task_done()来通知Queue当前task完成self.result_queue.put(answer)returnclass Task(object):def __init__(self, a, b):self.a = aself.b = bdef __call__(self):time.sleep(0.1) # pretend to take some time to do the workreturn '%s * %s = %s' % (self.a, self.b, self.a * self.b)def __str__(self):return '%s * %s' % (self.a, self.b)if __name__ == '__main__':# Establish communication queuestasks = multiprocessing.JoinableQueue() results = multiprocessing.Queue()# Start consumersnum_consumers = multiprocessing.cpu_count()print(('Creating %d consumers' % num_consumers))# 创建cup核数量个的子进程consumers = [ Consumer(tasks, results) for i in range(num_consumers) ]# 依次启动子进程for w in consumers:w.start()# Enqueue jobsnum_jobs = 10for i in range(num_jobs):tasks.put(Task(i, i))for i in range(num_consumers):tasks.put(None)#有几个进程加几个None，以此结束死循环# Wait for all of the tasks to finishtasks.join()# Start printing resultswhile num_jobs:result = results.get()print ('Result: %s' %result)num_jobs -= 1

思路：
类consumer ：进程对象，定义了进程怎么运行任务，怎么记录结果
类Task:定义了任务是什么：a*b=c
主程序：
声明2个队列：
1个是任务队列，类型：joinalbe queue(每get一个任务
执行完任务之后，都要调用一次self.task_queue.task_done()
表示这个任务完成)
2 结果队列，类型：Queue，存储任务的计算结果（10个数）
3 获取你机器的cpu个数
4 实例化10个Task类的实例，作为任务，放到任务队列
5 启动cpu个数的进程，依次start启动
6 多进程开始干这10个任务，每个任务的计算结果计入结果队列
7 任务队列名.join()这句话等待所有任务执行完毕
8 遍历结果队列，打印所有的结果

进程同步之：Lock

锁是为了确保数据一致性。比如读写锁，每个进程给一个变量增加 1 ，但是如果在一个进程读取但还没有写入的时候，另外的进程也同时读取了，并写入该值，则最后写入的值是错误的，这时候就需要加锁来保持数据一致性。

>#encoding=utf-8
from multiprocessing import Process, Lock
import time
def l(num,lock):  lock.acquire() # 获得锁time.sleep(0.2)print("Hello Num: %s" % (num))  lock.release() # 释放锁if __name__ == '__main__':  lock = Lock()  # 创建一个共享锁实例for num in range(10):  Process(target = l, args = (num,lock)).start()

进程同步之：Semaphore(多把锁)
Semaphore用于控制对共享资源的访问数量。Semaphore锁和Lock稍有不同，Semaphore相当于N把锁，获取其中一把就可以执行。可用锁的总数N在创建实例时传入，比如s = Semaphore(n)。与Lock一样，如果可用锁为0，进程将会阻塞，直到可用锁大于0。

#encoding=utf-8
import multiprocessing
import time
def worker(s, i):s.acquire()print(multiprocessing.current_process().name + " acquire")time.sleep(i)print(multiprocessing.current_process().name + " release")s.release()if __name__ == "__main__":# 设置限制最多3个进程同时访问共享资源s = multiprocessing.Semaphore(3)for i in range(5):p = multiprocessing.Process(target = worker, args = (s, i * 2))p.start()

最多只能有3个进程同时获得锁

进程同步之：event

Event提供一种简单的方法，可以在进程间传递状态信息，实现进程间同步通信。事
件可以切换设置和未设置状态。通过使用一个可选的超时值，时间对象的用户可以等待其状态从未设置变为设置。

#encoding=utf-8
import multiprocessing
import timedef wait_for_event(e):"""Wait for the event to be set before doing anything"""print('wait_for_event: starting')e.wait() # 等待收到能执行信号，如果一直未收到将一直阻塞print('wait_for_event: e.is_set()->', e.is_set())def wait_for_event_timeout(e, t):"""Wait t seconds and then timeout"""print('wait_for_event_timeout: starting')e.wait(t)# 等待t秒超时,此时Event的状态仍未未设置，继续执行print('wait_for_event_timeout: e.is_set()->', e.is_set())e.set()# 初始内部标志为真if __name__ == '__main__':e = multiprocessing.Event()print("begin,e.is_set()", e.is_set())w1 = multiprocessing.Process(name='block', target=wait_for_event, args=(e,))w1.start()w2 = multiprocessing.Process(name='block', target=wait_for_event, args=(e,))w2.start()#可将2改为5，看看执行结果w3 = multiprocessing.Process(name='nonblock', target=wait_for_event_timeout, args=(e, 2)) w3.start()print('main: waiting before calling Event.set()')time.sleep(3)# e.set()   #可注释此句话看效果
print('main: event is set')

两个等待的进程都继续执行

进程同步之：pipe（管道）

Pipe是两个进程间通信的工具。Pipe可以是单向(half-duplex)，也可以是双向(duplex)。通过mutiprocessing.Pipe(duplex=False)创建单向管道 (默认为双向)。一个进程从Pipe一端输入对象，然后被Pipe另一端的进程接收，单向管道只允许管道一端的进程输入，而双向管道则允许从两端输入。
Pipe的每个端口同时最多一个进程读写，否则会出现各种问题，可能造成corruption异常。Pipe对象建立的时候，返回一个含有两个元素的元组对象，每个元素代表Pipe的一端(Connection对象)。我们对Pipe的某一端调用send()方法来传送对象，在另一端使用recv()来接收。

#encoding=utf-8
import multiprocessing as mpdef proc_1(pipe):pipe.send('hello')print('proc_1 received: %s' %pipe.recv())#pipe.recv()会死等pipe.send("what is your name?")print('proc_1 received: %s' %pipe.recv())
def proc_2(pipe):print('proc_2 received: %s' %pipe.recv())#pipe.recv()会死等pipe.send('hello, too')print('proc_2 received: %s' %pipe.recv())pipe.send("I don't tell you!")if __name__ == '__main__':# 创建一个管道对象pipepipe = mp.Pipe()print(len(pipe))print(type(pipe))# 将第一个pipe对象传给进程1p1 = mp.Process(target = proc_1, args = (pipe[0], ))# 将第二个pipe对象传给进程2p2 = mp.Process(target = proc_2, args = (pipe[1], ))p2.start()p1.start()p2.join()p1.join()