多进程 multiprocessing（1） --python

环境 mac OS X EI Capitan

python3.5

实例

使用Process类

在multiprocessing中，进程的生成都是通过创建Process对象，例如

from multiprocessing import Process
def f(name):
print("hello", name)
p = Process(target=f,args=("world",))
p.start()
p.join()

在建立 Process 对象，通过

help(Process)

查看实例化Process时的初始化语法。

__init__(self, group=None, target=None, name=None, args=(), kwargs={}, *, daemon=None)

group:应该一直为None, 它的存在只是为了兼容threading.Thread

target:能被run()函数激活的可调用对象,通过查看源码可以看到，在run()函数里面

self._target = target``self._target(*self._args, **self._kwargs)

，也就是target是个函数，函数的参数形式

(*self._args, **self._kwargs)

arg：就是target函数的参数。

self._args = tuple(args)

,序列化args

kywargs:target函数的参数,

self._kwargs = dict(kwargs)``self._target(*self._args, **self._kwargs)

name：定义进程的名称，字符串类型，无语义意思，多个进程可能定义同样的名字。

daemon:True,Fasle,初始值为None,是否将进程设置为守护进程。此值在start()调用前必须赋值。

start():开始一个子进程

join()：主进程等待，直到所有子进程结束。

进程之间通讯

1 - Queues

是进程，线程安全的。

multiprocessing中的Queue几乎就是queues.Queues克隆。

from multiprocessing import Process, Queue

def f(q):
q.put([42, None, "helo"])

q = Queue()
p = Process(target=f, args=(q,))
p.start()
print(q.get())
p.join()

2 - Pipes

multiprocessing.Pipe([duplex]) 返回一对(conn1, conn2)代表通道两端的Connection 对象。

如果duplex为True (默认值)，通道是双向的，它们都有send(obj)，recv()。

如果duple为False,通道是单向的，conn1只能用来接受信息，conn2只能用来发送信息。

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()

⚠️两个线程或者进程在同一时间读写通道的同一端会导致数据丢失。当然进程在同一时间使用不同的端是不会有数据丢失的风险。

3 － 进程间同步

multiprocessing包含等同来自threading的所有同步原语。例如，可以使用一个lock使得一次仅仅一个进程打印标准输出(或者do something else)，目的就是为了避开多个线程使用同一资源。

from multiprocessing import Processing, Lock

def f(l, i ):
l.acquire()
print("hello world",i)
l.release()
lock ＝ Lock()

for num in rnage(10):
Process(target=f, args=(lock,num)).start()

4 - 进程间分享状态

在并发编程中，最好避免使用共享状态。尤其是在使用多进程。

当然如果有必要，multiprocessing提供了方法来实现这一操作。

(1)共享内存

通过使用Value,Array,数据可以存储在共享的内存映射中。

from multiprocessing import Process, Value, Array

def f(n, a):
n.value = 3.14
for i in range(len(a)):
a[i]=-a[i]

num = Value("d", 0.0)
arr = Array("i", range(10))

p = Process(target=f, args=(num, arr))
p.start()
p.join()

print(num.value)
#3.14
print(arr[:])
#[0, -1, -2, -3, -4, -5, -6, -7, -8, -9]

Value(typecode_or_type, *args, lock=True

Array(typecode_or_type, size_or_initializer, *, lock=True)

其中Array中的 ＊ 表示在 ＊之前的函数参数是位置参事，之后的函数参数是关键字参数，而且 ＊ 不能放在最后一个位置：

def f(*,x,y):
print(x,y)
f(2,8)
# TypeError: f() takes 0 positional arguments but 2 were given
f(x=2,y=8)
# 2,8

def f(x, *, y):
print(x,y)
f(3,4)
# TypeError: f() takes 1 positional argument but 2 were given
f(3,y=4)
#3,4

typecode_or_type可设值图表



(2)Server process

通过manager()创建一个manager对象能够控制具有python对象的server process。通过manager()创建的manager对象支持 list, dict, Namespace, Lock, RLock, Semaphore, BoundedSemaphore, Condition, Event, Barrier, Queue, Value and Array 等类型。

from multiprocessing import Process, Manager
def f(d,l):
d[1] = "1"
d["2"] = 2
d[0.25] = None
l.reverse()

with Manager() as manager:
d = manager.dict()
l = manager.list()
p = Process(target=f, args=(d,l))
p.start()
p.join()
print(d)
#{0.25: None, 1: '1', '2': 2}
print(l)
#[9, 8, 7, 6, 5, 4, 3, 2, 1, 0]

一个单独的manager可以被在同一个网络中的不同计算机多个进程分享。

l = manager.list([sequence])#创建list对象，并返回一个代理

进程池

pool类表示工作进程池，通过一些不同的方法，可以将任务从池中转移出来变成工作进程。

from multiprocessing import Pool
def f(x):
print(x*x)
return x*x
pool = Pool(processes=4)
result = pool.apply_async(f,[100])
print(result.get())
print(pool.map(f, range(10)))

以上进程池的执行使用的是异步的方式，进程循环运行过程中不等待apply_async的返回结果，在主进程结束后，即使子进程还未返回整个程序也会就退出。虽然 apply_async是非阻塞的，但其返回结果的get方法却是阻塞的，result.get()会阻塞主进程。可以在主进程中使用pool.close()与pool.join()来防止主进程退出。join()方法一定要在close或terminate之后调用。

pool.close()
pool.join()

以上是多进程使用的一个流程。其中每一块的更多语法，类方法，可以在官网的相应位置查找，或者查看源码。

参考：

multiprocessing — Process-based parallelism

https://docs.python.org/3/library/multiprocessing.html

https://pymotw.com/2/multiprocessing/basics.html

https://pymotw.com/2/multiprocessing/communication.html