Python并行编程(十二)：进程同步

1、基本概念

多个进程可以协同工作来完成一项任务，通常需要共享数据。所以在多进程之间保持数据的一致性就很重要，需要共享数据协同的进程必须以适当的策略来读写数据。同步原语和线程的库类似。

- Lock：一个Lock对象有两个方法acquire和release来控制共享数据的读写权限。

- Event：一个进程发事件的信号，另一个进程等待事件的信号。Event对象有两个方法set和clear来管理自己内部的变量。

- Condition：此对象用来同步部分工作流程，在并行的进程中，有两个基本的方法，wait()用来等待进程，notify_all用来通知所有等待此条件的进程。

- Semaphore：用来共享资源，比如：支持固定数据的共享连接。

- RLock：递归锁对象，其用途和方法同Threading模块一样。

- Barrier：将程序分成几个阶段，适用于有些进程必须在某些特性进程之后执行，处于Barrier之后的代码不能同处于Barrier之前的代码并行。

2、测试用例

使用barrier函数来同步两个进程

import multiprocessing

from multiprocessing import Barrier, Lock, Process

from time import time

from datetime import datetime

def test_with_barrier(synchronizer, serializer):

    name = multiprocessing.current_process().name

    synchronizer.wait()

    now = time()

    with serializer:

        print("process %s ----> %s" %(name, datetime.fromtimestamp(now)))

def test_without_barrier():

    name = multiprocessing.current_process().name

    now = time()

    print("process %s ----> %s" %(name, datetime.fromtimestamp(now)))

if __name__ == "__main__":

    # create a barrier and lock.

    synchronizer = Barrier(2)

    serializer = Lock()

    # create four processes

    Process(name='p1 - test_with_barrier', target=test_with_barrier, args=(synchronizer, serializer)).start()

    Process(name='p2 - test_with_barrier', target=test_with_barrier, args=(synchronizer, serializer)).start()

    Process(name='p3 - test_without_barrier', target=test_without_barrier).start()

    Process(name='p4 - test_without_barrier', target=test_without_barrier).start()

运行结果如下：

test_with_barrier函数调用了barrier的wait()方法，当两个进程都调用wait()方法时，他们会一起继续执行。

3、进程之间管理状态

Python的多进程模块提供了在所有的用户间管理共享信息的管理者(Manager)，一个管理者对象控制着持有Python对象的服务进程，并允许其他进程操作共享对象。

管理者特性：

- 它控制着管理共享对象的服务进程

- 它确保当某一进程修改了共享对象之后，所有的进程拿到的共享对象都得到了更新。

代码示例：

import multiprocessing

def worker(dictionary, key, item):

    dictionary[key] = item

    print("key = %d value = %d" %(key, item))

if __name__ == "__main__":

    mgr = multiprocessing.Manager()

    dictionary = mgr.dict()

    jobs = [multiprocessing.Process(target=worker, args=(dictionary, i, i*2)) for i in range(10)]

    for j in jobs:

        j.start()

    for j in jobs:

        j.join()

    print("Results:",dictionary)

运行结果：

上述代码创建了一个管理者字典dictionary，在n个job之间共享，每个job都会更新字典的某一个index，所有的job完成之后，最后打印该字典，所有数据均存在。

巴特西

Python并行编程(十二)：进程同步

最新文章

热门文章