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

杨利霞

爬虫（七十）多进程multiprocess（六十一）
! p3 H1 W3 u1 ?* Z/ v" u+ _

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

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

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

• 
  
  import os
  ( k5 A) t' w+ [" Y( S2 W$ S
  
• 
  $ H% K. q2 _4 m* S9 E# P
  import multiprocessing
  ) s& I1 F: r) G5 V
  1 J1 v/ f( V! |5 b3 o
• 
  
  # t7 P& @" F1 `* {9 l2 U
  
  3 O/ \; ]9 O+ n8 d$ v# e
• 
  
    Y2 O7 L! z% S; _" M5 ~) T% qdef foo(i):
  / Z: O2 Q' w3 e* ]
  " |2 f" B) a& h9 J* l5 h! N5 B
• 
  
  + ], J4 m* P" h1 M3 K    # 同样的参数传递方法
  
  
• 
  
  $ V9 c2 I! r8 }    print("这里是 ", multiprocessing.current_process().name)
  
  8 Y/ m+ n- B  i  W
• 
  
      print('模块名称:', __name__)
  
  
• 
  ( F5 C; h* ?/ _$ s! @
      print('父进程 id:', os.getppid())  # 获取父进程id
  5 p) e' D! b; w' i
  
• 
  # q( }' a) Y% F0 m' B/ m
      print('当前子进程 id:', os.getpid())  # 获取自己的进程id
  ! f+ ]- v  ?. Y1 r; q8 _* E
  
• 
  1 m& U* I8 b5 b
    Q5 X& L% Y: O2 Y    print('------------------------')
  
  
• 
  8 T) g& ?+ R$ D( c2 \
  
  8 f% o" C$ v, g( n
  * \" ]& z# t# N: \/ C
• 
  8 M8 K% n; z3 l
  if __name__ == '__main__':
  
  
• 
  
  
  ; g* Z6 `6 |' J/ D% [1 v+ e
  
• 
  % |, Z- ^9 R9 e
      for i in range(5):
  
  / A6 n; q! c% \& C" a) g8 }, y
• 
  - w* Q7 F" j& ?
  - A1 V7 M8 T4 P# T6 A        p = multiprocessing.Process(target=foo, args=(i,))
  3 H: J7 e/ Q# c, M6 K/ Z
  
• 
  % O3 V0 W! |$ i* f
          p.start()
  
  
  ; D! p& n) ~* ?- M+ a* R) F! {1 j8 R

运行结果：

• 
  
  2 t+ q% E# Q; [4 Q. e; o这里是  Process-2
  
  6 t# t1 D0 Y" y) e
• 
  + |0 E( S) _5 e% J
  : n( U4 f& b5 ]- G模块名称: __mp_main__
  + |4 |- I% E' i9 @; Y# N  l
  * [% _9 [- A. W6 l7 O0 \# [
• 
  & @' t' L. d1 D" y6 C
  父进程 id: 880
  
  6 T3 w% a2 M$ m7 u9 @
• 
  
  9 @: a5 k0 Y8 V6 p6 B  R$ u/ \当前子进程 id: 5260
  
  - Z( \- K1 w+ Y
• 
  9 Y9 J/ R$ y6 K' e: X" L
  : f) }& p5 G! E: S5 a4 j) a--------------
  0 R' u! H- K$ y8 P8 _9 E; Q
  ; u, ^6 {8 ^8 o6 S$ X* `) S) f) b7 X
• 
  
  ( W2 N. ~# h, q- ?# {这里是  Process-3
  3 U  S9 l" l: c) _( g4 h
  ( _* v9 i$ C" h! g" {; T
• 
  
  1 V/ A: z7 l, z( H. o6 j6 Y模块名称: __mp_main__
  # J  C2 f) U1 F8 l
  
• 
  6 M, t" M3 Q3 A
  5 v/ @- f! d1 v" J" m父进程 id: 880
  
  ; ^$ }+ n) [( r5 ^* Q
• 
  
  当前子进程 id: 4912
  
  
• 
  / F' K. p7 A/ |6 ^
  7 s" C& `& h- Q/ T--------------
  3 M" t% U/ z3 d0 |! I1 k
  
• 
  ( n0 d+ A( j$ G" M% D
  5 n7 F( b* z) W# s* n: b  R8 @这里是  Process-4
  5 m$ L5 X. n* ?' n' [" k
  3 r. h- J7 a2 y/ N
• 
  
  模块名称: __mp_main__
  3 N  H6 |7 X, H9 O5 Y. j1 @
  
• 
  
  父进程 id: 880
  7 J5 K& [5 H7 ^7 |* }) X0 \- z  k
  / A7 s; ~; m* E, N. D# Q* w# t
• 
  
  当前子进程 id: 5176
  
  ! O5 O* {+ u0 ^# Y' C( H
• 
  + \  @0 K2 D5 j5 i& ?
  --------------
  
  
• 
  + t1 ~1 S7 e1 n/ N- v4 ^
  7 N! T- _4 H; x+ ?3 U5 S这里是  Process-1
  ) {4 \) h% V4 h" m
  
• 
  : e+ T5 k! h9 j" W: ~
  2 ?# Q2 w- L: J0 j# X8 s模块名称: __mp_main__
  
  
• 
  
  父进程 id: 880
  
  
• 
  " b8 u- \( s0 k; G9 V
  & ~* h! j! N+ V1 U' C7 H当前子进程 id: 5380
  
  
• 
  " G; ]: i8 E1 I/ M) a+ R% T8 d
  --------------
  9 o/ o. A) v$ I  Q
  - B6 I6 |! x; `' I" b; j9 ?
• 
  4 ~$ H) ?! k; s% D
  这里是  Process-5
  
  8 n/ [/ _2 t, G) ~4 }/ j- C1 i9 u
• 
  
  模块名称: __mp_main__
  . y, D9 G/ i. C5 A2 P2 v+ n
    y# g! A  y/ T6 ^* o3 F
• 
  
  父进程 id: 880
  5 Z4 X" m1 j7 n0 T1 H# ~2 T$ o
  ' [: ~5 F# e, h% D& e& p
• 
  . ^5 y' e1 ~% {+ O8 Z. R2 g9 _! p8 T- }
  当前子进程 id: 3520
  ) U  s# L4 k6 ~/ c5 V2 p! m! v; j0 u
  & A" |7 h, ]; m; B4 l# y% A& R' O" |
• 
  
  --------------
  
  4 Y' H- }" t3 K2 _9 D1 b0 I
    u  k# M7 j: ]/ T% ^

1. 进程间的数据共享

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

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

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

• 
  
  9 {; o7 s' u6 b* P$ ~, hfrom multiprocessing import Process
  
  6 c! R4 ?2 _1 L9 j! P0 b4 e& c# Z
• 
  3 E0 H- d- L+ t, v; v
    d% M* I* h% w! w) s+ m
  / o9 N( {3 ?. G: `+ n9 }! Y( p+ p
  
• 
  : O$ |* t; ]/ V
  lis = []
  3 n5 J% |; d/ p+ L9 K" m' D/ W
  9 ]; \4 a  p& Z* p% l& X: i" i) P
• 
  : H' V! X3 D) H6 z& T: u: q1 m7 \$ f
  & B. k7 f. J) r$ y8 o
  
  8 f4 s' t% F+ R7 G! e6 V
• 
  
    h: z: P! G. D4 K: e! Odef foo(i):
  
  / R% P# X. x4 K/ V: T+ G+ H
• 
  
      lis.append(i)
  5 \1 S' u% b( P6 {) U# a# l7 I
  
• 
  " K# Q" B, B- G: H8 |
  ) K; i# d. |9 z. C+ s+ v7 j/ U    print("This is Process ", i," and lis is ", lis, " and lis.address is  ", id(lis))
  $ @6 A: g3 ~; H* C' `0 T
  
• 
  
  9 e( ]8 g9 n. z+ b3 h6 h7 q
  6 p/ @/ ^# w3 B
  : a9 G/ o4 ^2 R
• 
  
  4 g1 `, T" R7 E% t0 kif __name__ == '__main__':
  , A0 c; t( @! Z5 {
  
• 
  9 ?7 V3 p- S/ w; k9 n
      for i in range(5):
  
  ( p3 O* P+ }; A1 Z9 s/ |
• 
  
          p = Process(target=foo, args=(i,))
  % I4 ^/ l/ r2 q* V2 f8 Q
  % L% E# P, x# }  P5 M! l
• 
  
          p.start()
  
  % [6 G5 ?! j+ F% p7 i& i6 K
• 
  4 O/ K) L$ b! `0 E; p
  ; L8 W0 n! t) B% x( s4 ]    print("The end of list_1:", lis)
  " f) _+ K3 w  f6 U: U
  
  0 \% L4 h" b- j# N$ h/ @9 y: }  ]

运行结果：

• 
  . v0 f2 R6 O, ]6 l
  The end of list_1: []
  % f) l) F% D9 M
  
• 
  
  This is Process  2  and lis is  [2]  and lis.address is   40356744
  
  
• 
  
  6 [+ j& o% N( CThis is Process  1  and lis is  [1]  and lis.address is   40291208
  
  
• 
  
  ' `6 g6 f0 B. w4 i6 u1 s  K9 {. bThis is Process  0  and lis is  [0]  and lis.address is   40291208
  ) ~! w! g" Y8 p5 Q
  . j4 l. m" s6 e1 k  r
• 
  5 a- D! k3 [- H- p3 ]7 j- p
  , C( D0 w  @2 [. ?/ D/ o$ I( qThis is Process  3  and lis is  [3]  and lis.address is   40225672
  
  
• 
  9 N' ?% ?- p! Y! ~4 }5 D5 B1 K
  This is Process  4  and lis is  [4]  and lis.address is   40291208
  
  ; p) g/ r* ?! o8 x5 K
  ! N: u% l9 T) {  ~2 w. O% C

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

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

1.1 使用Array共享数据

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

• 
  3 ~" ~) r# O$ C( k# G% x$ }
  'c': ctypes.c_char, 'u': ctypes.c_wchar,
  # p2 `  Q3 {& b2 U
  / U! e' `/ f% q9 O
• 
  
  - I1 E) e  B$ ~; D6 y" n  a8 R'b': ctypes.c_byte, 'B': ctypes.c_ubyte,
  & D, c1 J# o9 Q( f. t; k( Z
  / O! f! A/ D. b
• 
  $ k2 v, d& V# _) i5 \/ O: G
  $ s) C4 ?  @) A, L% c+ j+ R, S) k! W'h': ctypes.c_short, 'H': ctypes.c_ushort,
  : |4 f7 ?6 ?( n2 d2 H/ D( E
  4 [8 g0 R+ @# d% Z6 q- u
• 
  
  'i': ctypes.c_int, 'I': ctypes.c_uint,
  0 l1 ]* M' b' J) E8 R
  5 S! A6 a% s" ]
• 
  
  1 J* N! P  j5 `. m6 t'l': ctypes.c_long, 'L': ctypes.c_ulong,
  , L3 r$ r( A5 ~6 K0 R' A3 @6 E
  " l. R' d1 r' L/ U, C
• 
  
  / r* w$ |$ k5 l' M& H, m'f': ctypes.c_float, 'd': ctypes.c_double
  
  5 Y6 `$ @) E% K5 y9 O1 C. F/ _
  

看下面的例子：

• 
  5 D1 }, q, @. h
  8 t$ n6 x! _' n5 z3 o" M/ P9 afrom multiprocessing import Process
  
  * Z( R1 b6 _7 O: R+ I
• 
  
  0 R. R& Z9 d' e$ t9 Gfrom multiprocessing import Array
  
  
• 
  
  
  ! y1 J5 e  p7 m& Q8 t  q, N4 R* G
  ; V/ O+ s8 Q( D
• 
  % a5 M" E$ }2 P5 y" v3 \, w
  5 c, Y+ r# K1 ~( j! ~( ldef func(i,temp):
  
  % W5 m1 g' T# {( X( @: P( s; Z
• 
  5 _& s4 Y3 {3 T7 N# ~
      temp[0] += 100
  , G0 R( c  r* K' G4 h* p
  % C+ E1 _1 d+ {9 d, B& K
• 
  # |, s  Z. h: n  D% P0 U
  4 g" S5 X  R0 N; ?- k' e% C    print("进程%s " % i, ' 修改数组第一个元素后----->', temp[0])
  
  
• 
  7 H4 q! t4 o! M5 b8 q
  * t7 A$ z) V- l, F% g- x. X# E
  & o9 X. D5 f; E
  
• 
  : w. R5 ~% T* g% s) U
  if __name__ == '__main__':
  * |# ]; o- M8 ~. p& z. @
    I  G1 g* R/ Q7 Z
• 
  
  ! v  d( e# O8 L" n. d, Q+ G9 Q    temp = Array('i', [1, 2, 3, 4])
  1 b# d& C" r/ X/ P
  
• 
  % w* i6 k+ ]- u0 [6 c* R7 }
  ' U0 f! x+ r) a3 X" {2 V    for i in range(10):
  : u; v3 e* i2 D% U9 p
  5 ]% j2 b! P% c0 B# \3 L
• 
  $ p8 O7 P  n2 U- n* H
          p = Process(target=func, args=(i, temp))
  ) }; }5 R( x- @
  
• 
  1 ]2 N( M. e0 O9 F7 V  ^+ G
  6 i5 a9 y; L! w) u( n1 \7 d        p.start()
  " {1 I7 D, D6 R; P
  ) A4 j0 Z* l) |& C
  + E3 A& b: M( m+ Q5 K

运行结果：

• 
  
  ) Z5 d8 b- }( r! U% N$ G7 _进程2   修改数组第一个元素后-----> 101
  , j9 U  R6 s8 w5 Q
  
• 
  ) U5 ]5 P# Q1 s
  + H6 ^. E! i8 M+ u1 b进程4   修改数组第一个元素后-----> 201
  
  
• 
  
  / C: P% s: {( Y进程5   修改数组第一个元素后-----> 301
    A3 M  e/ Z5 \. B- ^# f& h
  . c$ j- o9 f8 h/ T1 l$ \" y  [. e+ k
• 
  9 A3 M1 k9 y! ]0 ?, n& p% S
  3 j- v1 O% D7 S8 I) C& Z& Q进程3   修改数组第一个元素后-----> 401
  6 l* L" l" b$ O  x$ X! ^- ], {
  ) g3 b! q7 |/ @1 w6 C
• 
  
  9 n: l* D- j$ S6 [. R2 H% E5 Y- G5 I进程1   修改数组第一个元素后-----> 501
    g7 A; L2 S& M' W: h5 U' I' N
  
• 
  
  % }; T( N% M, @2 |进程6   修改数组第一个元素后-----> 601
  $ e% A  @; \; y* v3 f; Q
  
• 
  
  : p2 `+ B& I7 `+ T  {, p进程9   修改数组第一个元素后-----> 701
    H$ m% M$ m) P0 l2 J  V, O& R
  
• 
  ; P' Y+ A" ~  a! n# H
  进程8   修改数组第一个元素后-----> 801
  9 l1 }" y  X, |, j
  
• 
  
  进程0   修改数组第一个元素后-----> 901
  
  - E# ?4 y$ R) f/ u; t8 Q8 S
• 
  
  进程7   修改数组第一个元素后-----> 1001
  6 ?, ~5 F; {" S4 J; V3 X4 `6 q
  
  / p5 y4 w* c! \' M7 p; F7 b

1.2 使用Manager共享数据

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

• 
  
  5 r" w* J# L! Y( afrom multiprocessing import Process
  9 N( l  Y$ u2 k* @6 P- L3 M, x
  ! X7 j2 p# Q  k( {6 e5 `
• 
  
  - |& o. v" @$ G3 ifrom multiprocessing import Manager
  
  0 w$ K0 X- Y2 s3 X" K  ]
• 
  
  2 t' Q4 v9 F4 v" ~# S
  ' |( M  C( V1 c5 C- q2 Z  ?
  , [; k" u6 ^4 _# o
• 
  7 _( p! G, [+ o1 F+ c4 c) U$ |/ m* q3 e/ t
  def func(i, dic):
  
  ! j4 e" w+ b* N: m* ^* b; V
• 
  7 ]( }9 k; B8 x/ e+ K( D
      dic["num"] = 100+i
  % c5 q# ~  a; f# w! R* a  D- v1 J
  
• 
  0 _; w% Z$ _/ M1 z) o
  / T, r; W8 R( b4 w! t( c    print(dic.items())
  
  " k! o  I% t& }9 y2 z  T
• 
  
  1 d8 e6 E& v5 p
  
  $ y8 Y5 {! F9 W( m5 Q  F( }
• 
  + n3 y0 c, g( ?# N) k* A' h7 t
  if __name__ == '__main__':
  
  " k8 e' H) p& R
• 
  6 g2 J+ T  E% ~- p1 m
  % p9 T& Z! x/ S$ @/ b+ D    dic = Manager().dict()
  
  
• 
  / B6 M5 C5 X7 w" a
  3 A. g( E0 A! L1 V    for i in range(10):
  
  5 s5 h4 a: f- S) o5 a
• 
  
          p = Process(target=func, args=(i, dic))
  
  ( h9 i' T; r3 F; O' q
• 
  ' Q3 Q/ ^) E4 t/ S6 ~) _
          p.start()
  
  
• 
  
  # K5 c# H) d0 E  r; e        p.join()
  
  + X4 f- L9 r) M6 k% k& u
  0 k8 P6 C- G* x) g% |) S

运行结果：

• 
  
  ; c5 S: k2 m! }  Y8 K) r3 g" r+ n[('num', 100)]
  / I9 Y' {, w4 _0 G
  
• 
  * N4 ?' U) l) V1 C) P3 c7 R- v
  [('num', 101)]
  4 K0 U! s0 O# k1 O- P( M, S$ M
  ! z2 s( E- Z. p) H3 f
• 
  
  [('num', 102)]
  
  
• 
  
  [('num', 103)]
  6 W' B5 o8 r* V6 N6 D  x- [; a) D  j
  
• 
  # S% a$ ~4 V0 ]" h5 k
  [('num', 104)]
  ( V1 k$ @; b) m2 E% C+ U, n
  
• 
  
  5 R* z1 r' h* i5 _[('num', 105)]
  + N/ ~' r  b; x, Y# _" K- i
  / q4 h: y) M- j/ n
• 
  
  [('num', 106)]
  
  8 ^0 {. u. [4 C' v
• 
  5 R4 j$ N# y" r2 B, w' {; e) q+ Y6 d
  [('num', 107)]
  
  / S' p8 `# [  D$ Q' \
• 
  
  - p; V0 t& r  u% U7 s# I5 P[('num', 108)]
  
  # ?+ ]$ ^& V5 S" i+ w
• 
  
  ! U( K2 ~) z# r[('num', 109)]
  ) D2 p/ K* f/ |0 ^1 X/ N
  ; u0 ~# E5 F+ |
  

1.3 使用queues的Queue类共享数据

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

• 
  4 X9 B0 |5 {- ]( a# ]) _5 F3 B
  import multiprocessing
  6 @, I7 h/ c$ ^/ t" m) ^% f% E: g
  
• 
  
  from multiprocessing import Process
  , K+ }( R7 ]1 v7 {! _
  0 ^4 Y1 N# ^: m
• 
  
  from multiprocessing import queues
  $ M/ K. _8 g0 {( }
  ; ^; y1 h2 M' |0 v, Q8 b9 N
• 
    Z- k2 i( A3 u3 Y% Z# A9 u9 }
  , j" {" I4 n% M4 d3 i- N( F3 L
  3 W$ ?4 E  e- F
  
• 
  
  def func(i, q):
  
  ; r7 y3 X5 P0 z- v; [
• 
  3 |4 W& O7 j! |9 p! X
      ret = q.get()
  ; f& r2 o1 E. ]" Z0 e- T- A
  * G' m* F5 O8 ?
• 
  ) N7 |8 w; L, a
  # D, w, {1 I2 X) _' ~" F, P    print("进程%s从队列里获取了一个%s，然后又向队列里放入了一个%s" % (i, ret, i))
  " w' A4 n2 X* m& c* T
  
• 
  + I, P$ t; F4 ^2 R9 F
      q.put(i)
  " N' k" ~( k; o0 S- I
  
• 
  
  * `+ Q5 f9 n# B% Q( h9 k% s
  ! ?8 i2 u' V3 }6 r# s! C: i. ^
  
• 
  
  ! p0 D. b& g7 \2 p# _if __name__ == "__main__":
  0 Q$ m( a' x+ ^; \8 Z; z$ o- |6 T
  ; a9 k" q# ^9 @- I, B8 |
• 
  ; `) e7 F, P5 i- {- A6 l2 ?5 |
      lis = queues.Queue(20, ctx=multiprocessing)
  
  * k& W( u3 T* i, n! G, x9 B
• 
  
      lis.put(0)
  / X- U8 ?! u. @& s- ^0 V% E. ?
  ) X6 Y$ X2 Y+ y
• 
  
      for i in range(10):
  
  , }9 e* m$ n. B( q
• 
  
          p = Process(target=func, args=(i, lis,))
  - k' ^) U# B3 A9 z: b( h
  
• 
  / q7 j9 \5 F& j, `2 ~
  5 |9 H+ x) I5 A        p.start()
  6 n& |4 _! j  ^9 N: s
  
  

运行结果：

• 
  
  , u( S: T0 f. S7 e* o/ h8 C进程1从队列里获取了一个0，然后又向队列里放入了一个1
  : o; w) L  J1 f" t8 h6 |( t1 Y8 B
  7 C9 L3 N$ @* K8 l9 C" u, [
• 
  3 V5 S3 m7 R, Y' x- _( s5 X; ]
  进程4从队列里获取了一个1，然后又向队列里放入了一个4
  
  ) u# f9 H' l# ~) @/ A
• 
  
    U& u# X* J8 t) S6 j) w进程2从队列里获取了一个4，然后又向队列里放入了一个2
  
  % J- ~1 O# e: V9 U
• 
  
  ; c" o( B7 r, q, }* ]1 n进程6从队列里获取了一个2，然后又向队列里放入了一个6
  
  
• 
  9 D- l! K6 Y: n9 F+ C0 J
  进程0从队列里获取了一个6，然后又向队列里放入了一个0
  
  - A8 s( V& e4 W+ |, G2 V
• 
  8 Y' ^, a% p  g2 v
  / s) V6 v. x3 _1 c* j进程5从队列里获取了一个0，然后又向队列里放入了一个5
  & t% V% x) Z: H( M: T
  
• 
  ' S3 ]: R6 x5 R& O
  ' k* W' m2 \/ `进程9从队列里获取了一个5，然后又向队列里放入了一个9
  * V8 s& T4 W7 x
  & R  s$ }9 _( A  E( s
• 
  
  进程7从队列里获取了一个9，然后又向队列里放入了一个7
  
  2 e4 N$ Y- Z1 C) _+ h# [( A) l
• 
  
  进程3从队列里获取了一个7，然后又向队列里放入了一个3
  % b* e# q( a9 K4 p0 y/ L
  
• 
  
  ; I) v+ S, `; e% o! a( A进程8从队列里获取了一个3，然后又向队列里放入了一个8
  
  
  

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

2. 进程锁

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

• 
  
  from multiprocessing import Process
  
  0 O$ s7 i# S' w. H
• 
  
  5 e0 Z3 L/ h2 N& S) T! _from multiprocessing import Array
  
  ( |8 G* v$ t/ w- y2 b- o5 p
• 
  6 k" y) b2 m3 i2 t
  6 j4 W" y4 g2 b9 J4 Lfrom multiprocessing import RLock, Lock, Event, Condition, Semaphore
  " ?. N, c! ^" F8 u) R. e$ {
  , B+ P- y2 {' \: E' k
• 
  ; N& @0 T; _+ J  q
  import time
  
  
• 
  
  ! P8 L5 v  b  q2 ~+ L
  
  6 v. g5 s9 E% N2 O) L; M$ ]; d1 m
• 
  
  + Y  i  \9 I/ |* |/ ]; @& r. Ydef func(i,lis,lc):
  8 |# ]* o$ k2 v$ [
  8 ]0 t- y! j$ ?/ ]7 X# U. ^, b) g
• 
  
  + k/ T+ d, n) Y% c+ H8 i    lc.acquire()
  
  
• 
    r7 D, f. A) L, e
      lis[0] = lis[0] - 1
  
  
• 
  
  , e$ g) k* _& F( d2 l    time.sleep(1)
  4 U3 v% g2 l9 s3 m
  . @+ G" M1 T1 t! P7 p$ ^9 B" Y
• 
  9 e8 V, U! u' Y4 w3 K
      print('say hi', lis[0])
  
  1 f, b) g- L0 O# J
• 
  
  ( X; n% `5 w: @9 b# d( z( [' p2 L    lc.release()
  
  " o6 U. j  v, J& g4 N
• 
  
  2 [# i4 @9 N. p1 i  J2 m: t% l
    `* \, E) R9 a: A3 `
  
• 
  
  # [( M; [' T$ g9 s9 Qif __name__ == "__main__":
  
  ! S$ ^7 H% ?0 L, \/ R# _- \6 \
• 
  # }' E, n) M: d- i$ r, j8 |1 K
  " o! ^! O$ w' H0 [( d8 I' D9 B    array = Array('i', 1)
  
  
• 
  
  ; T4 w: c$ N, e1 P  c5 M& i% R    array[0] = 10
  
  " A4 g: `) o2 N
• 
  ) }  X# \# |. Z) R- U9 Y( p6 I$ a
  % X9 w6 X( y% Q% S- E$ N4 P    lock = RLock()
  
  / @$ n: G" ^' {# F
• 
  
      for i in range(10):
  
  
• 
  
          p = Process(target=func, args=(i, array, lock))
  
  8 C/ }) Q1 M- q  f) }
• 
  & t' {5 p+ V$ I$ v$ M- r5 |
  * N  U% l; W$ [        p.start()
  6 u2 h* B; X3 k6 w" e
  
  $ L: g& c4 x7 Y( l! [  }$ c

运行结果：

• 
  
  say hi 9
  
  
• 
  
  say hi 8
  4 j* O6 n1 N/ C: d+ P5 ?
  $ B6 j) x% m* [
• 
  
  say hi 7
  
  9 l, Q# b1 X0 r- `# r1 F2 u
• 
  
  say hi 6
  
  
• 
  
  say hi 5
  $ k* s5 v% K- m: I  V
  7 ~9 {0 e* n+ r7 W7 e' n
• 
  
  say hi 4
  
  % R/ k3 N7 ^* D& G( w
• 
  $ e7 a8 {: l; s4 Z8 y
  say hi 3
    E; |( ?+ w3 D4 ?* V! r2 i9 p
  
• 
  4 X1 G% ~9 i% Y/ s
  say hi 2
  , H' k1 }- X$ W  s+ w+ f
  : Z' C( W8 Y. {+ o
• 
  
  * g2 f' j% X# w: X2 j: r) ksay hi 1
  , y0 [- ]: q. c$ y
  
• 
  4 K5 S0 _  N3 {, A6 |0 [  p
  0 ^$ @' g" T0 m6 g% d$ wsay hi 0
  
  ) p( K- Y- e: |, ?# h/ e. n5 Y: Z$ {
  

3. 进程池Pool类

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

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

进程池中常用的方法：

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

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

• terminate() 立刻关闭进程池

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

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

  
  

• 
  
  from multiprocessing import Pool
  
  
• 
  , B  I& M0 _$ Y/ x" ]" r* {$ t' i
  import time
  & R4 ]7 j$ q: D
  
• 
  
  
  
  
• 
  - E1 O; {) n6 t) x% c& }' Y  A
  + a- T; B& T* t  q* i. bdef func(args):
  
  
• 
  3 |4 o5 A$ V6 ~2 l3 W& D" T# _
      time.sleep(1)
  9 d3 Z: w& y4 D$ h8 l
  ( O4 Q5 |* I4 e. ~; @4 x
• 
  
  4 H5 C4 a* f  W0 F2 |1 m8 t" _    print("正在执行进程 ", args)
  
  
• 
  
  
  
  , M5 E: y2 L3 g* O/ C
• 
  
  if __name__ == '__main__':
  7 c+ Z/ A! N' T% e
  " F- {& ]$ ]! i) ?7 I5 U8 T
• 
  , z3 n" H0 d) p9 i
  7 z  z! [3 I8 x  P0 ?
  
  / U( L* O  R: T1 o9 }
• 
  ; Z. n- l2 {" B6 @
  7 n& h* t1 n* B' |* `! X2 K4 f2 }    p = Pool(5)     # 创建一个包含5个进程的进程池
  
  $ A- q2 P5 T1 U5 Y6 l
• 
  
  ) ^5 B1 Q, b1 c1 B/ F0 H
  * o8 y; d) Q- H$ A# O* o
  
• 
  
  0 I- g/ N$ q, [: F    for i in range(30):
  
  
• 
  5 ]) V' J! O1 d& a) m2 V2 V% ?
          p.apply_async(func=func, args=(i,))
  # }. F0 f# t( g; U# X* f1 c
  
• 
  . P: T0 v. M+ x
  9 @- V3 X5 N; Z( _4 f8 f
  + \& L$ T! E( R
  ) d: x* y, _+ s7 c5 d+ v
• 
  & P0 Z5 d9 K* a4 C$ D- K3 i
      p.close()           # 等子进程执行完毕后关闭进程池
  
  2 \# T  w- B6 C% H" J9 a2 \
• 
  
      # time.sleep(2)
  
  
• 
  $ F6 X/ o5 u( r" E; L6 j
      # p.terminate()     # 立刻关闭进程池
  
  $ `" y7 w  D: N0 R7 G/ `& n. h
• 
  
  $ O  {- Q  |8 p- b$ n) p    p.join()
  ( [5 b: q2 t  N
  2 j( K( J+ W0 K$ ]2 @
  

: D8 k( z6 ?2 V9 z1 G8 C2 F2 Z% B

请继续关注我

- t1 H' c# i! q5 T$ Z