爬虫（七十）多进程multiprocess（六十一）

杨利霞

爬虫（七十）多进程multiprocess（六十一）

Python中的multiprocess提供了Process类，实现进程相关的功能。但是它基于fork机制，因此不被windows平台支持。想要在windows中运行，必须使用if __name__ == '__main__:的方式，显然这只能用于调试和学习，不能用于实际环境。

另外，在multiprocess中你既可以import大写的Process，也可以import小写的process，这两者是完全不同的东西。这种情况在Python中很多，请一定要小心和注意。

下面是一个简单的多进程例子，Process类的用法和Thread类几乎一模一样。

• 
  
  import os
  
  - c+ Q* k: P4 F+ j  h
• 
  7 g: @7 [% L4 I, `1 G' R9 A
  : d0 c1 m" D9 m2 F' o; Fimport multiprocessing
  
  
• 
  - x8 `, S* g7 L4 z; a' u
  9 p, N1 d  r) g$ e1 V
  
  
• 
  
  def foo(i):
  ) z) N# d- u0 D( q8 g' g# I& Y
  
• 
  
      # 同样的参数传递方法
  + e' a. C  o9 \$ n7 x
  . e. y/ w5 P! M, S8 q! o' ~( t) m" s
• 
  
    K4 w% y; u# {/ x    print("这里是 ", multiprocessing.current_process().name)
  : x8 `: d. Y" @# I
  
• 
  0 N! n( l2 W: K2 K' O
    d' X; @3 s6 U! u    print('模块名称:', __name__)
  
  / M3 |2 z) E2 F
• 
  ! d% I3 S9 U- \6 F8 D  G" H
      print('父进程 id:', os.getppid())  # 获取父进程id
  
    ~' R4 I* I5 L& r+ I: }+ d
• 
  4 c; v" F; k/ e' {% M6 S
  7 [8 N/ s' w* ~# {, ?, N7 T    print('当前子进程 id:', os.getpid())  # 获取自己的进程id
  
  5 A3 ?5 X9 n: g% C7 j% u1 n
• 
  & M1 m5 X. w- ~
  + M+ j  V# {" i& W* E    print('------------------------')
  $ D) n/ M3 G$ h3 v$ Y) S' Y7 `
  ) Y8 {- t; w+ Q$ e  V
• 
  2 C6 \9 n) F; g
  & {0 U# Y- o- ]2 e0 D! P3 ^8 u, \
  
  
• 
  
  2 d$ R& `* O5 {1 ~if __name__ == '__main__':
  
  
• 
  ( u, d; o8 k) |! l
  + K! Z0 S7 b9 V* q
  2 F; i& w0 G" ?% N8 O# e! S7 l% U
  / q) l- ~+ T- Z; ~) W
• 
  
      for i in range(5):
  9 u; |7 I7 }) I( t' G- z
  
• 
  
          p = multiprocessing.Process(target=foo, args=(i,))
  
  
• 
  8 x9 w7 b2 }2 G0 A0 \% R8 \
  0 e3 h: n/ ^" X: g. ?5 _        p.start()
  
  5 W3 }4 Q, K# O
  6 M: e# j0 D/ O2 `1 y! T3 a

运行结果：

• 
  
  : H/ P; u; n. N. v这里是  Process-2
  
  . n: U7 ^/ h2 `$ K9 Y1 d" ~( O
• 
  
  模块名称: __mp_main__
    i% D  k' v) D0 N5 ~9 c
  3 \. J# P4 O4 }  `0 p* h! \( x
• 
  0 f1 N0 `$ b" e6 n
  父进程 id: 880
  
  ( x! |: }  g6 u
• 
  - V  q: j4 q# Q3 C
  6 w+ v5 O! m0 S当前子进程 id: 5260
  
  , j) t& Z8 ?4 Q( |* U; R: [
• 
  
  --------------
  
  
• 
  
  6 ]# i2 W" F7 g& U  q6 j  X这里是  Process-3
  
  1 l# q$ t/ N# q& c6 z
• 
  + v. P* y8 k2 M$ j$ k5 \6 E- B6 T
  模块名称: __mp_main__
  
  0 q( o4 C$ |+ w9 J- {9 k% t7 D
• 
  
  父进程 id: 880
  
  * U5 x/ f2 }1 R' y0 T+ m
• 
  ; F3 h& t- V: `( x7 e9 Y: [4 _$ k
  $ I3 {+ u5 q; ~' M" n1 y当前子进程 id: 4912
  - y3 r3 s( D( D; u' R  H: U9 K. _
  
• 
  
  7 B9 R# |6 e4 O+ x5 `3 C--------------
  + D1 C) C! K2 l- w) s6 f& v
  
• 
  , X6 _" w! N4 G" a1 x2 G
  " O( _7 n( g  p$ V7 k  J这里是  Process-4
  ! L; A8 I5 x: U  ?
  0 D  B& k( c9 e5 R
• 
  ! H4 _9 v, l5 t0 c" _. k( m7 {
  模块名称: __mp_main__
  
  
• 
  
  父进程 id: 880
  / Z+ A8 p6 @# }% y+ Y
  
• 
  
    Z3 j) u0 C$ C% O% z+ |当前子进程 id: 5176
  
  5 C2 F$ Y: h  r6 e3 {2 `
• 
  
  --------------
  
  
• 
  
  ' b: a. Z3 j: G0 e) Y# w这里是  Process-1
  
  ; p0 o! V9 [: Y+ o( c& _
• 
  
  + j: o2 }* v4 U& R$ a模块名称: __mp_main__
  6 [% [4 w9 v9 E: s
  
• 
  
  父进程 id: 880
  
  5 _( J  d, f7 g% Y, m: I
• 
  
  + }, }8 ?0 @+ a- L: t当前子进程 id: 5380
  
  4 A9 I' V* U! _) \5 u# P  n
• 
  1 {% |' W2 f0 {- J  l9 v
  ; v. [8 k6 U2 m0 u8 C1 O( x# T--------------
  
  
• 
  
  ! t( U2 i& _* K( T这里是  Process-5
  
  4 J2 `8 T! I  s0 C( o8 X; h
• 
  
  ; H1 C2 j0 g: E+ P/ U模块名称: __mp_main__
    W; l& o! r. o& p
  
• 
  ' s- P/ f% Q. W/ u7 w/ ~$ e
  父进程 id: 880
  
  6 n" a# P* X/ ^  _: G4 W# b
• 
  % h2 k& Z* g' S) ~
  , A0 ], T. I$ x; Z当前子进程 id: 3520
  
  . v  m: p- s* h0 H2 r7 u
• 
  8 f4 l& B5 @* s
  # {$ u0 T3 k+ o) _--------------
  0 H( ~# i: v  \% R8 j2 ~. s/ J
  
  

1. 进程间的数据共享

在Linux中，每个子进程的数据都是由父进程提供的，每启动一个子进程就从父进程克隆一份数据。

创建一个进程需要非常大的开销，每个进程都有自己独立的数据空间，不同进程之间通常是不能共享数据的，要想共享数据，一般通过中间件来实现。

下面我们尝试用一个全局列表来实现进程间的数据共享：

• 
  ) q& }* h: K* d) F( O
  from multiprocessing import Process
  
  
• 
  , \, z9 b) [) b0 l: E. Z3 z
  ! Z) z4 R& y+ f3 k3 o. H" }0 q2 X9 o
  ( ?! J% j, z8 T2 d
  ) ^9 L: w/ s, O4 r: [
• 
  
  lis = []
  
  & r' e2 i; s2 v: m
• 
  1 U! h& Z) [; F2 q) U& y
  2 f! U9 s  |- o2 V
  
  : y" l! W5 p; G6 T, A- n
• 
  
  ( }+ ?$ F3 u( R% Z/ n. _/ X. ~def foo(i):
  % c4 Z5 A2 l  Q* H" E' r: E
  
• 
  $ i+ }1 Y7 j* y3 Y) w% r+ f+ N9 t; |
      lis.append(i)
  1 C3 v- ~1 p& o9 ~' ~
  
• 
  
      print("This is Process ", i," and lis is ", lis, " and lis.address is  ", id(lis))
  
  2 Q' l; K7 w0 p
• 
  
  2 p% s3 p8 I/ B
  6 N. o- u6 a* \
  
• 
  - ]- ]% @3 x7 D) t; `2 R7 s/ b: e
  if __name__ == '__main__':
  
  
• 
  . P- j: O  E5 P* ^) V1 s
      for i in range(5):
  
  " \8 C/ @# t; w# e* C2 ?5 l! ]9 [
• 
  , I" G. L5 Q+ M' y5 s
          p = Process(target=foo, args=(i,))
  , S, t2 ^9 h( N6 F# r
  & T) \3 G, u$ F. m5 H
• 
  2 s1 |1 L% w4 t
  ) ~' }- p% q2 {* R  a2 s$ I* M$ {' Y        p.start()
  8 F3 }& D" v( M" y$ P" U* o- G" x
  ' \8 x/ z1 ^9 X0 q, H! g) L2 H
• 
  
      print("The end of list_1:", lis)
  
    z$ o# g' `7 Q/ o( a* _
  / U9 w9 c  H* O6 U0 R! O

运行结果：

• 
  : L* j( w. v5 O( m4 o5 E( n& f) K
  : m( D2 K, ?8 D! cThe end of list_1: []
  
  
• 
  0 `& C' R  Z6 a, V6 t6 A3 J7 L
  4 M. {6 D3 H3 TThis is Process  2  and lis is  [2]  and lis.address is   40356744
  
  
• 
  9 H3 L! }% y9 j6 q" h
  This is Process  1  and lis is  [1]  and lis.address is   40291208
  
  
• 
  ' d1 l2 Z2 f+ c9 I
  This is Process  0  and lis is  [0]  and lis.address is   40291208
  
  6 H! `: @- {' y3 Y# v
• 
  
  . e4 h9 d$ f) UThis is Process  3  and lis is  [3]  and lis.address is   40225672
  
  
• 
  
  & y) P4 N( l& F; [$ Z" ^% ?This is Process  4  and lis is  [4]  and lis.address is   40291208
  
  7 b- d9 U2 D/ K5 T3 c
  - y4 q! z$ x& d

可以看到，全局列表lis没有起到任何作用，在主进程和子进程中，lis指向内存中不同的列表。

想要在进程之间进行数据共享可以使用Queues、Array和Manager这三个multiprocess模块提供的类。

1.1 使用Array共享数据

对于Array数组类，括号内的“i”表示它内部的元素全部是int类型，而不是指字符“i”，数组内的元素可以预先指定，也可以只指定数组的长度。Array类在实例化的时候必须指定数组的数据类型和数组的大小，类似temp = Array('i', 5)。对于数据类型有下面的对应关系：

• 
  
  , d* T4 N0 w% B* \4 o4 M( P# N'c': ctypes.c_char, 'u': ctypes.c_wchar,
  
  5 b; P8 b6 e7 s& ?
• 
  
  $ B1 ~) S. e; J( ?' \: H'b': ctypes.c_byte, 'B': ctypes.c_ubyte,
  # h- L- _0 G, [6 J4 E0 }+ ~
  
• 
  
  , T, C+ o7 q' {'h': ctypes.c_short, 'H': ctypes.c_ushort,
  
  ' j, o" p, s  a: e0 |  N8 C! E
• 
  
  # n+ w( [% W- ~; V- p& ['i': ctypes.c_int, 'I': ctypes.c_uint,
  
  
• 
  
  'l': ctypes.c_long, 'L': ctypes.c_ulong,
  
  ! ~! X! u+ _7 C+ R
• 
  4 E& A: m  V) \: c; T+ N/ L: F) _
  'f': ctypes.c_float, 'd': ctypes.c_double
  3 w) W) l, K; {. x
  " `, ~3 B" x2 W' ~/ w! n/ }
  

看下面的例子：

• 
  2 a- P* t! y- Y" u6 N0 [
  2 Y# K/ o  V% t8 ^from multiprocessing import Process
  5 f% q7 K5 r7 f; C- D
  
• 
  / n# V4 y, T( y
  & O5 h8 I% ^5 u8 pfrom multiprocessing import Array
  # n' E- Q; s& E
  
• 
  
  , J) m1 T1 S2 J8 X2 {& Z
  4 c) u$ |5 U4 o4 d8 d: W  C3 N
  # D( z2 k4 e) I$ r
• 
  " Q  p$ j! u' A# L: L) m6 f
  ) a+ j& u2 V2 qdef func(i,temp):
  
  
• 
  
  ! e2 R  C: {$ C/ p    temp[0] += 100
  4 ]- ]8 q, D' i9 v8 n
  ; k; o, i/ z0 z4 k
• 
  4 S! d0 r7 }- m4 J5 ]+ q# y
      print("进程%s " % i, ' 修改数组第一个元素后----->', temp[0])
  " g) {$ y6 d7 q7 ?' |# T: e# A
  9 m% q1 w5 {5 P& r' e* `/ N
• 
  ' U! p* U9 t$ o' [8 v) ~
  ( ^! F6 {4 @) k, h$ ^
  & h1 V4 P& L3 f) k: V5 `
  % y7 w+ P2 J/ }* Y. j$ [# \
• 
  
  - _7 b. c0 m3 Sif __name__ == '__main__':
  ) ^& t& `# g: G! |
  
• 
  
      temp = Array('i', [1, 2, 3, 4])
  3 c; N4 _1 X9 O
  
• 
  
  ! [7 x+ R& y* P7 z: _6 E- T' X    for i in range(10):
  6 q: O* W. A: O2 Z
  , K: t! |) b4 n7 k2 J
• 
  
  " i9 j3 v2 P: k: ?2 P% h: p" ^        p = Process(target=func, args=(i, temp))
  8 a3 ^( f. j( s4 J
  
• 
  
  3 h/ B9 J/ b% c& `+ d% i! q        p.start()
  
  
  ( T% m! N+ ^8 ^$ T; u

运行结果：

• 
  
  进程2   修改数组第一个元素后-----> 101
  
  4 q% V4 q- O3 u, |! w6 @
• 
  
  7 x- E7 e; B: q) I" L/ F  t! ~进程4   修改数组第一个元素后-----> 201
  
  / C. j# R: d) @$ \9 x
• 
  
  1 [8 W: n) c" M3 y进程5   修改数组第一个元素后-----> 301
  
  & W2 R/ _" V* }8 |% J1 S, G
• 
  , _- X, N7 o$ s
  4 L( C8 W/ ^! K: @/ M% l" F1 g进程3   修改数组第一个元素后-----> 401
  5 @7 k* a: S2 J7 l! p7 U
  
• 
  
  进程1   修改数组第一个元素后-----> 501
  / \0 i2 L# e  q% d- w( `
  
• 
  
  4 a/ }2 ]/ p, B# m% {& k进程6   修改数组第一个元素后-----> 601
  
  
• 
  
  进程9   修改数组第一个元素后-----> 701
  
  
• 
  
  ! U) |; X: Z5 ~) B; f, v进程8   修改数组第一个元素后-----> 801
  ( `+ `+ V: d* l( U
  : V. F% v. A( w5 {% |0 ^
• 
  
  进程0   修改数组第一个元素后-----> 901
  ; g- c6 x9 D2 ^/ `. W
  
• 
  / \' R! U9 ?/ v" I
  进程7   修改数组第一个元素后-----> 1001
  
  - J: E5 Y2 o6 V( d* Q: T
  - @* q. z0 F- k6 U0 K

1.2 使用Manager共享数据

通过Manager类也可以实现进程间数据的共享。Manager()返回的manager对象提供一个服务进程，使得其他进程可以通过代理的方式操作Python对象。manager对象支持 list, dict, Namespace, Lock, RLock, Semaphore, BoundedSemaphore, Condition, Event, Barrier, Queue, Value ,Array等多种格式。

• 
  
  ) E  F5 X* d* K* _& Sfrom multiprocessing import Process
  $ Y; }, R1 l4 d* a* a; I+ t0 V+ P
  $ O1 ~* G( g  Q, T& w! z
• 
  & I! b+ f) ?2 \# `0 T
  ( d  e3 z# x' p# F: ffrom multiprocessing import Manager
  & g/ |% M; I! z" \+ X/ |1 [/ G
  % Q3 N( T8 M( _! i
• 
  
  
  2 Y5 k" z) m5 ]9 _  O
  " ?# j4 y5 E# b% h# c4 B4 |
• 
  
  def func(i, dic):
  / _% S. r4 r2 \+ ^2 h0 S( }
  . U+ C8 S4 {9 |- d
• 
  " Q" K" L- j" k! _: s6 }% n# L& [. R
  5 g. \, \5 O: j) @7 h    dic["num"] = 100+i
  " y; I3 M: [0 u7 d
  7 ?  U+ S8 T7 ^* T, m2 F. {
• 
  
      print(dic.items())
  $ E9 A! Y9 \! ^/ }
  
• 
  
  ! a2 O4 S* r6 h- P; M
  3 I" \/ g8 g5 Q9 Y$ H/ ^" j
  
• 
  
  if __name__ == '__main__':
  
  
• 
  9 e0 G1 N- u" L
      dic = Manager().dict()
  ) U; f1 l2 z) t, i' b
    B7 t4 h5 y  A, J3 t" h* p
• 
  " ]: B8 }/ G2 a9 i+ ]- y
      for i in range(10):
  $ j, T  [4 g! ^5 Y0 p+ W8 u
  
• 
  7 l! f/ V2 j+ j, o9 i! r" l
          p = Process(target=func, args=(i, dic))
  
  1 U: d  T6 Y1 V# d8 M) V# s
• 
  
  0 D+ p, `- E! `( b8 W        p.start()
  
  5 |! [/ M4 i( w
• 
  ) x4 z1 c! c, {0 z% L' Y/ K
          p.join()
  
  
  

运行结果：

• 
  
  9 y% @, N* B* n) A0 t[('num', 100)]
  ; h) g  h% M! Q0 _1 o5 J8 u
  ; f% y( B0 Q0 @
• 
  ; ~, h5 p* ]7 g. d7 d0 U
  [('num', 101)]
  * o) a- U" A  g5 K1 D5 m
  
• 
  
  - P3 Y( B$ @3 o. j' {, i6 P[('num', 102)]
  ; x% u& ?: A( e) ], U, x
  
• 
  3 o% k' a5 B1 U, z
  [('num', 103)]
  
  + D2 r" W5 @% C5 d6 H* {; u# [
• 
  1 L2 ]$ h' t: t$ S$ y
  [('num', 104)]
  
  - h  e6 s9 M" Q8 s- Q4 x; x* e: r
• 
  
  6 j  H' m) Z7 T( O; b8 C! N[('num', 105)]
  
  
• 
  : D* _1 t; _) A
  + @. ]* g0 a5 \7 j% F[('num', 106)]
  
  ! m8 ]2 \) Z( Y% b" _6 F
• 
  ' h) h( c! _9 z! Y' B
  - u8 H! {( ^/ D$ y4 L[('num', 107)]
  - g) X& N# i+ f# ^
  
• 
  + M+ s8 F$ v$ E
  [('num', 108)]
  / [( N5 c: O$ E1 U- Y
  
• 
  
  [('num', 109)]
  
  # U; c9 U% z2 L3 m3 o8 n8 r" M2 o9 n% C8 Q
  ; m& Y' r3 Y" ?9 l+ K1 g. a1 X7 l

1.3 使用queues的Queue类共享数据

multiprocessing是一个包，它内部又一个queues模块，提供了一个Queue队列类，可以实现进程间的数据共享，如下例所示：

• 
  ! O" {2 K+ D% L* x! |, h! t' F
  $ h! e& v& W7 himport multiprocessing
  9 }  G) O% T7 R  l
  " K& a  M" Q6 f/ [
• 
  ) _* V1 x4 N  g! S4 T% w
  from multiprocessing import Process
  
  
• 
    m( T! o  q' z/ q- n2 D/ g
  from multiprocessing import queues
  
  % B" g' w6 H! Y& K5 Y8 _( z
• 
  0 ~6 ?2 Y; P( ~- ~" w* s& _
  
    ?$ |) x( z& \/ h
  
• 
  
  def func(i, q):
  
  , B% Y7 V, u. w/ H+ P
• 
  
      ret = q.get()
  ; J+ M! i, j0 x9 `# C, E7 c
  
• 
  
  , Y7 v9 Q# j7 j% l6 g    print("进程%s从队列里获取了一个%s，然后又向队列里放入了一个%s" % (i, ret, i))
  7 @8 z" R- X- |9 s' _
  * O2 n8 A8 K8 |  ?) v! o( k/ v5 Z
• 
  ) d1 D9 Q$ @4 E) |$ k8 o6 d; j
  ) C$ u+ X4 O: t1 ?! v    q.put(i)
  # e; L; K* h/ E2 K! y9 j
  # ~" q! s' J" F+ @
• 
  
  
    w3 k2 r4 q' v9 |% X
  # l; f5 Y7 U" j
• 
  . n' v4 U' |8 m! n$ F
    m0 M- h. u7 j& k# z0 |0 Cif __name__ == "__main__":
  5 z" _1 s' l8 `5 y# e  ]& B# k( ?
  
• 
  # _3 J, }+ }5 r% @% _% b
      lis = queues.Queue(20, ctx=multiprocessing)
  
  2 i; n! w$ X$ M$ ]/ x9 F+ @  a2 L
• 
  
      lis.put(0)
  - I# f2 C( A7 |
  
• 
    s$ l& I( J9 B5 I' Q$ o
  & K4 a0 e6 c& j( Q( k6 K/ T    for i in range(10):
  
  
• 
  
          p = Process(target=func, args=(i, lis,))
  
  
• 
  , g4 e) \. p: h6 `. G1 M
  ( k" q. g. J0 y$ V- @" A8 a4 l        p.start()
  2 o! p* d! _' R& G+ n% t/ _. s' c" s+ E
  
  7 G7 ~2 A% K2 d" g: `% O

运行结果：

• 
  
  " h/ W0 l- V" P$ ?9 q进程1从队列里获取了一个0，然后又向队列里放入了一个1
  ; r' K2 w( l+ X/ F: W
  2 j9 ?# Q1 I! @) ?# j+ o5 h. q4 _
• 
  
  进程4从队列里获取了一个1，然后又向队列里放入了一个4
  % k# g; P, R4 F6 D1 B* P5 Y: P9 u
  3 \/ v, C! s$ |3 I; c# B2 S
• 
  + S# Q$ Z4 b& M- _2 Z% _
  $ N! V9 h) V8 ]9 M进程2从队列里获取了一个4，然后又向队列里放入了一个2
  % s$ }* \" D  \5 i9 d( l
  + F. A0 y8 P% Z; j! |2 S
• 
  / t" t9 u8 R0 ^7 y
  ( g2 c+ H$ M/ [. [9 |& e进程6从队列里获取了一个2，然后又向队列里放入了一个6
  
  : o8 q0 D8 T) e! O' H( Y
• 
  ' k  @5 D. K  y  W
  进程0从队列里获取了一个6，然后又向队列里放入了一个0
  7 n! I  U. G& j/ C9 ^$ a6 r
  
• 
  
  进程5从队列里获取了一个0，然后又向队列里放入了一个5
  $ [' l6 Q6 N0 \& p; l
  
• 
  # l: J. f  g: ]& r9 o8 s. e6 s
  * ?% g  @3 J% i/ l9 m进程9从队列里获取了一个5，然后又向队列里放入了一个9
  ) W/ B. u/ E8 z, d
  
• 
  ) B8 H' N. s6 K
  8 p* p4 w/ Z4 s( X% j/ g8 J进程7从队列里获取了一个9，然后又向队列里放入了一个7
  4 b3 [- Q. @, X: _  {2 R+ L
  9 a+ q. n0 Y4 J  k7 E
• 
  
  3 y+ _1 v* I: J' ]# _. k$ V- A: I, b进程3从队列里获取了一个7，然后又向队列里放入了一个3
  
  & F! r. M3 w; h0 O: U8 g3 n; a
• 
  
  进程8从队列里获取了一个3，然后又向队列里放入了一个8
  
  
  : U# p0 o! p$ q

关于queue和Queue，在Python库中非常频繁的出现，很容易就搞混淆了。甚至是multiprocessing自己还有一个Queue类(大写的Q)，一样能实现queues.Queue的功能，导入方式是from multiprocessing import Queue。

2. 进程锁

为了防止和多线程一样的出现数据抢夺和脏数据的问题，同样需要设置进程锁。与threading类似，在multiprocessing里也有同名的锁类RLock，Lock，Event，Condition和 Semaphore，连用法都是一样样的，这一点非常友好！

• 
  2 C/ C+ m' B. u! G: N. l& _7 {
  from multiprocessing import Process
  
  - i4 d4 f+ X. F* N% H$ I' ~  ]
• 
  + u$ V" g& l+ v1 q9 B
  from multiprocessing import Array
  3 t1 q' Q4 y3 h' d, h4 G' m
  5 c* a- g# J# R/ }) Y
• 
  
  ! ~$ J* t" e, e% `from multiprocessing import RLock, Lock, Event, Condition, Semaphore
  
  
• 
  
  " k" S0 n9 l: Pimport time
  / e; d/ p3 g& j: s4 |& `
  $ H+ X  z9 i: K
• 
  
  
  5 X- i. R9 U0 R7 S8 o6 S
  
• 
  
  def func(i,lis,lc):
  ' e4 v$ o) N+ t3 Q9 b0 X& A. y7 K
  ' ]2 {6 o& W" z% g
• 
  
  + A+ Y2 [6 ^+ b4 f6 F    lc.acquire()
  ' i) ^+ e* v. H8 J+ ^' U& B* D
  & N) Z" p2 t! B, v( {. ]+ a: N
• 
  , ^* H( p6 Y+ @5 o& D
  " y% D# g; g. s1 g& o/ z    lis[0] = lis[0] - 1
  
  ) n9 q! T  y& i$ p  k) a  w
• 
    `, |/ _- S6 X' e$ k! C6 v
      time.sleep(1)
  
  # l; B" e: H3 e. s' T. M
• 
  
      print('say hi', lis[0])
  
  
• 
  
  4 s2 B; e+ d8 ^! D8 M    lc.release()
  . H0 H9 V! H! w# s! h$ n
  8 S# l4 S& a2 [, L( f
• 
  
  
  0 f( y; k& F# L5 k
  ( n' ^; v( p) T( }3 z/ a
• 
  ( q/ d+ h' Q3 o: B$ M% |! d. Z0 h
  if __name__ == "__main__":
  
  + u3 p) N6 n$ s4 F: l2 e
• 
  
      array = Array('i', 1)
  . ]1 T3 g* x7 V! }1 s" ]) z6 [' U
  5 j! P3 O  Q( a0 k' ^$ ]* g
• 
  
  / u9 e" G" u$ x  n4 u7 }6 Y) f    array[0] = 10
  
  
• 
  
  & t' ?( ?/ S+ n0 x( j0 C0 V    lock = RLock()
  
  
• 
  
  . p$ {/ o/ T- [  Y* T    for i in range(10):
  " H- P3 w- m6 X$ m( K# ?
  
• 
  - W9 Q- l4 M  K' W+ q
  / |& c) y& Z- E4 G  W        p = Process(target=func, args=(i, array, lock))
  
  
• 
  
  5 P3 n! i$ r; w) L- K: r        p.start()
  " @$ s# N  K3 Y" O: K
  " {' i" N8 b0 m& c# G- Z, |. H
  1 V' f; P. V' e) A, S

运行结果：

• 
  
  8 E, J6 ?+ ^6 M: O8 xsay hi 9
  3 z1 @# r3 j) O: q+ w( R% C
  
• 
  ) H( t8 W) {: o) ~) ~; X. `
  # l, V/ e( [  s: N; ssay hi 8
  
  
• 
  , {. n+ ^2 C' P, K0 e
  9 m  f- E; a0 \& o4 wsay hi 7
  9 x5 ]) v! w% ]
  
• 
  
  say hi 6
  
  
• 
  ' c8 F3 n+ @" T0 t
  5 U+ R# `" W8 N5 g  {8 f1 M& Fsay hi 5
  " V1 Z, J. q& |+ f: h: ~0 H2 Z2 ~
  ( }# O3 q& j0 O: x& \1 }, R1 R& N5 N* W
• 
  ! U, q4 I4 D1 I* a* ?$ b# {) Q
  say hi 4
  / }- F) I; l' Q5 J8 @+ l
  
• 
  6 G( f0 ]. Z0 x  [' Q' I/ _& Q4 q
  say hi 3
  # E9 i2 `$ l$ f7 j- ?; e
  4 I) O- _5 Q2 [/ Z" Q
• 
  
  say hi 2
  
  
• 
  # T& |2 u$ }; B" N2 k
  say hi 1
  
  
• 
  3 c, W3 b6 d. B; E# @+ K
  % P3 j) Y# ]0 |4 qsay hi 0
  $ g7 Z8 k% I  c2 V8 }; Z
  
  

3. 进程池Pool类

进程启动的开销比较大，过多的创建新进程会消耗大量的内存空间。仿照线程池的做法，我们可以使用进程池控制内存开销。

比较幸运的是，Python给我们内置了一个进程池，不需要像线程池那样要自己写，你只需要简单的from multiprocessing import Pool导入就行。进程池内部维护了一个进程序列，需要时就去进程池中拿取一个进程，如果进程池序列中没有可供使用的进程，那么程序就会等待，直到进程池中有可用进程为止。

进程池中常用的方法：

• apply() 同步执行（串行）

• apply_async() 异步执行（并行）

• terminate() 立刻关闭进程池

• join() 主进程等待所有子进程执行完毕。必须在close或terminate()之后。

• close() 等待所有进程结束后，才关闭进程池。

  
  

• 
  ; s0 j7 \' A, E# M
  from multiprocessing import Pool
  4 _6 \. d  u$ e9 ~
  
• 
  " F4 S* \1 c! G
  import time
  , O, p" T* W. {5 X$ M4 V
  4 Y+ E' _0 V$ F, j; q
• 
  8 X/ F. l* z1 d: g* \
  7 Z. T3 T+ C( I- q; V
  
  
• 
  
  & D2 z' ]" Q4 _, Bdef func(args):
  
  
• 
  ! o, h: u6 Z' e1 u& W
  - R. @) t; L: s2 e4 A    time.sleep(1)
  ) q( K8 D' `( ]# \$ d
  
• 
  ! x; l! W  \3 f: {
  * p5 g' W2 Q; ^) z, q. j    print("正在执行进程 ", args)
  7 T/ M! D2 _' o4 ^
  
• 
  % {6 j. a( S& v+ C% _  A+ p
  , v% v1 F7 z4 e$ c, [: s! U& ^
  , ?2 [, A6 _' o( l0 s! {
  
• 
  - z" A" i1 L9 u# i. c5 i& S4 y1 x
  + \9 d3 N% U0 \4 {2 T  k7 Sif __name__ == '__main__':
  
  2 z% I- {% q, |9 \% {; W: G
• 
  
  * ]  q5 D- x9 I2 H
  
  * a2 Q0 \) ~' O1 v( R' U
• 
  
  7 q) [8 {& ~& n" i    p = Pool(5)     # 创建一个包含5个进程的进程池
  $ R  i6 g* X9 p0 c
  - s" g) y3 M5 ~* I
• 
  
  
  
  5 m# M7 t6 H0 K* ]$ {5 P
• 
  ( e3 i$ `! z' F  j; ]
  ; W* r* n4 X0 D/ d3 ]/ {    for i in range(30):
  * {) s# u; G. R, {% n/ Z
  3 n1 q" H* A& o8 i2 U( \0 X
• 
  
  $ c* ?/ Z, u8 Y! f; E% n        p.apply_async(func=func, args=(i,))
  
  
• 
  ( `* r+ c6 z$ x3 M4 d7 E- t
  9 F7 m( ^( q1 R; K
  
  6 b( L8 l! N- Y2 [
• 
  
  ) E( L* I6 Q) E0 O4 l% J# H+ W    p.close()           # 等子进程执行完毕后关闭进程池
  / S! T8 ]6 o! O5 R5 f$ V" \
  
• 
  0 j/ R  K0 k' X4 Y; E1 J
      # time.sleep(2)
  
  
• 
  9 t# F  e& G* w$ Z
  8 F( W# Q: w: j1 X    # p.terminate()     # 立刻关闭进程池
  4 t( ^& l4 }- Y/ O
  
• 
  , T1 ]- [& l  T& d0 E5 \+ J
  ' s( Z# C; ]6 L9 C# }    p.join()
  2 ]7 b0 Z  B& U* |1 t& j+ U
  
  ) F* x  C" N$ a

4 H% S8 @2 m- j4 @* C- w) ?: h

请继续关注我

2 [- k! L& B5 @( D
$ J( j, c6 x8 J* u$ v