python进程池详解

 

python进程池multiprocessing.Pool和线程池multiprocessing.dummy.Pool实例

1，Python的multiprocessing 实现多cpu 多进程

为啥用multiprocessing？

*如今cpu都多核了， 写大型程序你还在用单线程的传统写法？很不fashion并且效率很低， 你out了。 

*那用multi-threading不就结了？C语言能够， 编译完执行没问题。 可是python这种解释型的语言用多线程就不行了， python的multithreading效率并不高。听说是受制于GIL (global interpreter lock) 的锁机制, 该锁只能工做于单个cpu core。这样别的cpu core干着急也帮不上忙。

*那怎么办呢？ 本身fork一堆子进程，而后管理呗？ 呵呵，DIY？这不是从新造轮子吗。

*亲， 用python现成的multiprocessing库吧。支持多个cpu core，简直就是为了多核cpu打造的， python 2.6以后都支持。

不用说了，直接上干货

*sample 1， 建立process, target=f 函数名， args传递参数

# encoding: utf8
# @Time    : 2017/10/26 20:54
# @Author  : Ycs
# @Site    : 
# @File    : mymultiprocessing.py
# @Desc    : Python的multiprocessing 实现多cpu 多进程, python进程池multiprocessing.Pool和线程池multiprocessing.dummy.Pool实例

from multiprocessing import Process,Queue,Pipe,Pool
import time


def f(name):
    print('hello:', name)


if __name__ == '__main__':
    #建立process, target=f 函数名， args传递参数
    p=Process(target=f,args=('bbb',))
    p.start()
    p.join()

    #支持队列

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

if __name__ == '__main__':
    q=Queue()
    p=Process(target=ff,args=(q,))
    p.start()
    print(q.get()) #这里prints "[42, None, 'hello']"
    p.join()

    #支持Pipe

def f3(conn):
    conn.send([43,None,'hello'])
    conn.close()

if __name__ == '__main__':
    parent_conn,child_conn=Pipe()
    p=Process(target=f3,args=(child_conn,))
    p.start()
    print(parent_conn.recv()) # 这里prints "[42, None, 'hello']"
    p.join()

    #支持Pool， 雇一群worker来干活

def f4(x):
    return x*x

if __name__ == '__main__':
    pool=Pool(processes=4) # start 4 worker processes
    result=pool.apply_async(f4,(10,)) # evaluate "f(10)" asynchronously
    try:
        print(result.get(timeout=1))
    except TimeoutError:
        pass
    print(pool.map(f4,range(10)))
    it=pool.imap(f4,range(10))
    print(it.next())
    print(it.next())
    print(it.next(timeout=1)) # raises TimeoutError
    result=pool.apply_async(time.sleep,(10,))
    print(result.get(timeout=1))

 

注意事项:

* 在UNIX平台上，当某个进程终结以后，该进程须要被其父进程调用wait，不然进程成为僵尸进程(Zombie)。因此，有必要对每一个Process对象调用join()方法 (实际上等同于wait)。对于多线程来讲，因为只有一个进程，因此不存在此必要性。

* multiprocessing提供了threading包中没有的IPC(好比Pipe和Queue)，效率上更高。应优先考虑Pipe和Queue，避免使用Lock/Event/Semaphore/Condition等同步方式 (由于它们占据的不是用户进程的资源)。

* 多进程应该避免共享资源。在多线程中，咱们能够比较容易地共享资源，好比使用全局变量或者传递参数。在多进程状况下，因为每一个进程有本身独立的内存空间，以上方法并不合适。此时咱们能够经过共享内存和Manager的方法来共享资源。但这样作提升了程序的复杂度，并由于同步的须要而下降了程序的效率。

Reference:

[1] http://www.ibm.com/developerworks/aix/library/au-multiprocessing

[2] http://docs.python.org/2/library/multiprocessing.html

python中的多进程处理

 

1)Process
  要建立一个Process是很简单的。

1. from multiprocessing import Process
2.  
3. def f(name):
4.     print('hello', name)
5.  
6. if __name__ == '__main__':
7.      p = Process(target=f, args=('bob',))
8.      p.start()
9.      p.join()

  要得到一个Process的进程ID也是很简单的。

1. from multiprocessing import Process
2. import os
3.  
4. def info(title):
5.     print title
6.     print 'module name:', __name__
7.     print 'process id:', os.getpid()
8.  
9. def f(name):
10.     info('function f')
11.     print 'hello', name
12.  
13. if __name__ == '__main__':
14.     info('main line')
15.     p = Process(target=f, args=('bob',))
16.     p.start()
17.     p.join()

  建立进程：multiprocessing.Process([group[, target[, name[, args[, kargs]]]]])
  参数：
  group:    None,它的存在仅仅是为了与threading.Thread兼容
  target:   通常是函数
  name:     进程名
  args:     函数的参数
  kargs:    keywords参数

  函数：
  run()                  默认的run()函数调用target的函数，你也能够在子类中覆盖该函数
  start()                启动该进程
  join([timeout])        父进程被中止，直到子进程被执行完毕。
                         当timeout为None时没有超时，不然有超时。
                         进程能够被join不少次，但不能join本身
  is_alive()             
  terminate()            结束进程。
                         在Unix上使用的是SIGTERM
                         在Windows平台上使用TerminateProcess

  属性：
  name                   进程名
  daemon                 守护进程
  pid                    进程ID
  exitcode               若是进程尚未结束，该值为None
  authkey                    
            
2)Queue
  Queue相似于queue.Queue，通常用来进程间交互信息
  例子：

1. from multiprocessing import Process, Queue
2.  
3. def f(q):
4.     q.put([42, None, 'hello'])
5.  
6.  if __name__ == '__main__':
7.      q = Queue()
8.      p = Process(target=f, args=(q,))
9.      p.start()
10.      print(q.get())    # prints "[42, None, 'hello']"
11.      p.join()

  注意：Queue是进程和线程安全的。
  Queue实现了queue.Queue的大部分方法，但task_done()和join()没有实现。    
  建立Queue:multiprocessing.Queue([maxsize])
  函数：
  qsize()                             返回Queue的大小
  empty()                             返回一个boolean值表示Queue是否为空
  full()                              返回一个boolean值表示Queue是否满
  put(item[, block[, timeout]])       
  put_nowait(item)
  get([block[, timeout]])
  get_nowait()
  get_no_wait()

  close()                             表示该Queue不在加入新的元素
  join_thread()                       
  cancel_join_thread()

3)JoinableQueue
  建立：multiprocessing.JoinableQueue([maxsize])
  task_done()
  join()

4)Pipe

1. from multiprocessing import Process, Pipe
2.  
3. def f(conn):
4.     conn.send([42, None, 'hello'])
5.     conn.close()
6.  
7. if __name__ == '__main__':
8.     parent_conn, child_conn = Pipe()
9.     p = Process(target=f, args=(child_conn,))
10.     p.start()
11.     print(parent_conn.recv())   # prints "[42, None, 'hello']"
12.     p.join()

  multiprocessing.Pipe([duplex])      返回一个Connection对象

5)异步化synchronization
 

1. from multiprocessing import Process, Lock
2.  
3. def f(l, i):
4.     l.acquire()
5.     print('hello world', i)
6.     l.release()
7.  
8. if __name__ == '__main__':
9.     lock = Lock()
10.  
11.     for num in range(10):
12.         Process(target=f, args=(lock, num)).start()

6)Shared Memory

1. from multiprocessing import Process, Value, Array
2.  
3. def f(n, a):
4.     n.value = 3.1415927
5.     for i in range(len(a)):
6.         a[i] = -a[i]
7.  
8. if __name__ == '__main__':
9.     num = Value('d', 0.0)
10.     arr = Array('i', range(10))
11.  
12.     p = Process(target=f, args=(num, arr))
13.     p.start()
14.     p.join()
15.  
16.     print(num.value)
17.     print(arr[:])

1>Value
2>Array

7)Manager

1. from multiprocessing import Process, Manager
2.  
3. def f(d, l):
4.     d[1] = '1'
5.     d['2'] = 2
6.     d[0.25] = None
7.     l.reverse()
8.  
9. if __name__ == '__main__':
10.     manager = Manager()
11.  
12.     d = manager.dict()
13.     l = manager.list(range(10))
14.  
15.     p = Process(target=f, args=(d, l))
16.     p.start()
17.     p.join()
18.  
19.     print(d)
20.     print(l)

8)Pool

1. from multiprocessing import Pool
2.  
3. def f(x):
4.     return x*x
5.  
6. if __name__ == '__main__':
7.     pool = Pool(processes=4)              # start 4 worker processes
8.     result = pool.apply_async(f, [10])     # evaluate "f(10)" asynchronously
9.     print result.get(timeout=1)           # prints "100" unless your computer is *very* slow
10.     print pool.map(f, range(10))          # prints "[0, 1, 4,..., 81]"

multiprocessing.Pool([processes[, initializer[, initargs]]])

函数：
  apply(func[, args[, kwds]])
  apply_async(func[, args[, kwds[, callback]]])
  map(func,iterable[, chunksize])
  map_async(func,iterable[, chunksize[, callback]])
  imap(func, iterable[, chunksize])
  imap_unordered(func, iterable[, chunksize])
  close()
  terminate()
  join()
 

1. from multiprocessing import Pool
2.  
3. def f(x):
4.     return x*x
5.  
6. if __name__ == '__main__':
7.     pool = Pool(processes=4)              # start 4 worker processes
8.  
9.     result = pool.apply_async(f, (10,))   # evaluate "f(10)" asynchronously
10.     print(result.get(timeout=1))          # prints "100" unless your computer is *very* slow
11.  
12.     print(pool.map(f, range(10)))         # prints "[0, 1, 4,..., 81]"
13.  
14.     it = pool.imap(f, range(10))
15.     print(next(it))                       # prints "0"
16.     print(next(it))                       # prints "1"
17.     print(it.next(timeout=1))             # prints "4" unless your computer is *very* slow
18.  
19.     import time
20.     result = pool.apply_async(time.sleep, (10,))
21.     print(result.get(timeout=1))          # raises TimeoutError

    

9)杂项
multiprocessing.active_children()          返回全部活动子进程的列表
multiprocessing.cpu_count()                返回CPU数目
multiprocessing.current_process()          返回当前进程对应的Process对象
multiprocessing.freeze_support()
multiprocessing.set_executable()

10)Connection对象
send(obj)
recv()
fileno()
close()
poll([timeout])
send_bytes(buffer[, offset[, size]])
recv_bytes([maxlength])
recv_bytes_info(buffer[, offset])  

1. >>> from multiprocessing import Pipe
2. >>> a, b = Pipe()
3. >>> a.send([1, 'hello', None])
4. >>> b.recv()
5. [1, 'hello', None]
6. >>> b.send_bytes('thank you')
7. >>> a.recv_bytes()
8. 'thank you'
9. >>> import array
10. >>> arr1 = array.array('i', range(5))
11. >>> arr2 = array.array('i', [0] * 10)
12. >>> a.send_bytes(arr1)
13. >>> count = b.recv_bytes_into(arr2)
14. >>> assert count == len(arr1) * arr1.itemsize
15. >>> arr2
16. array('i', [0, 1, 2, 3, 4, 0, 0, 0, 0, 0])