多进程multiprocessing

from multiprocessing import Process

import os

def info(title):

    print(title)

    print('module name:', __name__)

    print('parent process:', os.getppid())

    print('process id:', os.getpid())

    print("\n\n")

def f(name):

    info('\033[31;1mfunction f\033[0m')

    print('hello', name)

if __name__ == '__main__':

    info('\033[32;1mmain process line\033[0m')

    p = Process(target=f, args=('bob',))

    p.start()

    p.join()

进程间的通讯

Queues(队列)

from multiprocessing import Process, Queue

def f(q):

    q.put([42, None, 'hello'])

if __name__ == '__main__':

    q = Queue()

    p = Process(target=f, args=(q,))

    p.start()

    print(q.get())    # prints "[42, None, 'hello']"

    p.join()

Pipes(管道)

from multiprocessing import Process, Pipe

def f(conn):

    conn.send([42, None, 'hello'])

    conn.close()

if __name__ == '__main__':

    parent_conn, child_conn = Pipe()

    p = Process(target=f, args=(child_conn,))

    p.start()

    print(parent_conn.recv())   # prints "[42, None, 'hello']"

    p.join()

通过Pipe()生成的两个连接对象，表示管道的两端。

每个连接对象都有 send() 和 recv()方法。

要注意的是：如果两个进程(或线程)，在同一时间尝试从管道的同一端读或写时，数据会被变脏。

当然，在同一时间，进程间使用不同的管道进行读写是没有任务风险的

managers（进程间共享资源）

from multiprocessing import Process, Manager

def f(d, l):

    d[1] = '1'

    d['2'] = 2

    d[0.25] = None

    l.append(1)

    print(l)

if __name__ == '__main__':

    with Manager() as manager:

        d = manager.dict()

        l = manager.list(range(5))

        p_list = []

        for i in range(10):

            p = Process(target=f, args=(d, l))

            p.start()

            p_list.append(p)

        for res in p_list:

            res.join()

        print(d)

        print(l)

support types list, dict, Namespace, Lock, RLock, Semaphore, BoundedSemaphore, Condition, Event, Barrier, Queue, Value and Array.**

Manager()返回的manager对象控制了一个server进程，此进程包含的python对象可以被其他的进程通过proxies来访问。从而达到多进程间数据通信且安全

实现共享内存和变量等。

proxies是什么？server process模型详解

在这个模型当中，有一个manager进程，负责管理实际的对象。真正的对象也是在manager进程的内存空间当中。所有需要访问该对象的进程都 需要先连接到该管理进程，然后获取到对象的一个代理对象（Proxy object），通常情况下，这个代理对象提供了实际对象的公共函 数 的代理，将函数参数进行pickle，然后通过连接传送到管理进程当中，管理进程将参数unpickle之后，转发给相应的实际对象 的函数，返回值（或者异常）同样经过管理进程pickle之后，通过连接传回到客户进程，再由proxy对象进行unpickle，返回给调用者或者抛出 异常

可用于不同计算机之间的资源共享

manager和proxy之间的连接可以是基于socket的网络连接，也可以是unix

pipe。如果是使用基于socket的连接方式，在使用proxy之前，需要调用manager对象的connect函数与远程的manager进程建立连接。

由于manager进程会打开端口接收该连接，因此必要的身份验证是需要的，否则任何人都可以连上manager弄乱你的共享对象。mp库通过

authkey的方式来进行身份验证。

在实现当中，manager进程通过multiprocessing.Manager类或者BaseManager的子类实现。

BaseManager提供了函数register注册一个函数来获取共享对象的proxy。这个函数会被客户进程调用，然后在manager进程当中执行。

这个函数可以返回一个共享的对象（对所有的调用返回同一个对象），或者可以为每一个调用创建一个新的对象，通过前者就可以实现多个进程共享一个对象。

进程同步

from multiprocessing import Process, Lock

def f(l, i):

    l.acquire()

    try:

        print('hello world', i)

    finally:

        l.release()

if __name__ == '__main__':

    lock = Lock()

    for num in range(10):

        Process(target=f, args=(lock, num)).start()

lock加锁保证进程间共享资源的一致性

进程池

from  multiprocessing import Process,Pool

import time

def Foo(i):

    time.sleep(2)

    return i+100

def Bar(arg):

    print('-->exec done:',arg)

pool = Pool(5)

for i in range(10):

    pool.apply_async(func=Foo, args=(i,),callback=Bar)

    #pool.apply(func=Foo, args=(i,))

print('end')

pool.close() # 必须在join之前close，否则会报错

pool.join()#进程池中进程执行完毕后再关闭，如果注释，那么程序直接关闭。

apply

主进程会阻塞于函数。主进程的执行流程同单进程一致

apply_async

非阻塞的且支持结果返回后进行回调