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

杨利霞

3 d1 ]4 D% O+ h0 u' z  f爬虫（七十）多进程multiprocess（六十一）
& @; @7 |! F9 P/ q, E) r& ?

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

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

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

• 
  
  import os
  
  
• 
  / r) ]% D$ U& H5 k# q2 y
  4 H7 ~, }  ?) T  m; D0 Aimport multiprocessing
  
  3 J0 X3 b4 a5 v+ x- a' M; d
• 
  
  + m) I/ u7 ]; ?. y
  
  
• 
  7 i* f2 ^0 U7 Q4 v
  : h: f! h8 ]3 R' G7 G. Z7 `+ Wdef foo(i):
  * X9 d  i2 X! N" G6 L) x+ P, c9 H! o' t4 _$ M
  4 M1 ]# ^) F+ L! [1 q
• 
  $ r) G) d$ P0 L$ J6 r
      # 同样的参数传递方法
  
  5 X3 Y, f4 D! D( c# E5 L' S
• 
  0 p2 d5 x/ ?0 v. r' J
      print("这里是 ", multiprocessing.current_process().name)
  : S& A7 @6 a, E4 W1 t
  * k3 V% _, u2 h7 l% C; O; I$ N0 o
• 
  
  9 ?( ], b. V1 |/ R# b    print('模块名称:', __name__)
  
  ; V; d6 s+ i7 g5 b$ p
• 
  & ~* c, Z9 C! n5 O, |
  2 s2 o  L) o9 N8 S; c5 m( l# b7 s    print('父进程 id:', os.getppid())  # 获取父进程id
  , L# K9 K, X; U8 U, u8 {$ T1 P
  
• 
  
  0 E$ d) Y2 N2 R: h1 R, x6 ^  u" l    print('当前子进程 id:', os.getpid())  # 获取自己的进程id
  . K' f" Q# z3 X" k' f
  
• 
  
      print('------------------------')
  
  ; Z+ W6 S( K& C" T1 I2 w
• 
  % y5 k$ K0 u( w; Z" w2 |% W
  
  
  ' }/ t. `: f& W' i
• 
  $ S% R. c- N. l% L- X
  ) \; W: d/ i+ Z+ lif __name__ == '__main__':
  - |2 a: {, b% u( e0 I. B
  : }' n$ u. @# _. ?$ e) m
• 
  # M" T2 |9 Z# S2 p
  
  + f& W# J( x  U' \# L. v2 d: V
  4 s2 E# j: ]" A
• 
  
  / C- \' G* b4 k, r    for i in range(5):
  + @- z- y' _& u. k
  
• 
  6 U9 Y" b( ?, b
  " z- y% }) W' q1 G! _7 j        p = multiprocessing.Process(target=foo, args=(i,))
  * u6 `( ^5 h3 j% l+ j" q1 J5 J9 v% |
  8 R& z, G! I* O+ {. V
• 
  
  , H3 C) p3 R: D# b: y        p.start()
  + V  B' C  N' g, X( v! W
  * M# O$ z, S, m8 @" _
  

运行结果：

• 
  3 c; B7 i& ~5 F+ g
  * g# p/ r+ ~5 k这里是  Process-2
  7 ~9 \& h, _, j! G$ v
  / i( k! `9 V: i+ h8 ~! z2 d
• 
  
  7 i- X, c+ s  ~# [, l1 n0 r* F4 m* J模块名称: __mp_main__
  - i- F* u  Z' t
  
• 
  
  父进程 id: 880
  : x' U) I' g' `2 S
  ' q7 w) E" M" ]9 y$ ]" p
• 
  
  当前子进程 id: 5260
  ' u  y+ t- `! y
  
• 
  
  --------------
  ; q8 i2 r; D3 ]6 n- M
  
• 
  
  ; x. o) f7 G0 {, |8 t这里是  Process-3
  % E/ w. Y8 P# F0 ?2 \: @/ Z% M
  6 ^' H4 ~3 p. U8 ^( l4 n+ x
• 
  0 y2 f- U. T* v' Y, W! j! u9 S, D+ e
  模块名称: __mp_main__
  ' C5 l3 U6 R. \& g# n
  
• 
  . H; {  ]1 W  \: c- b) w$ Y5 P9 y
  : H  B0 }! E8 x! ~! F* A父进程 id: 880
  
  
• 
  
  7 v+ A5 C$ o5 K  O6 S  s3 ?* Q当前子进程 id: 4912
  
  # j8 L$ W. Y3 W4 \9 T1 B  V; E4 d
• 
  1 x4 ^6 z6 k$ r8 `
  --------------
  1 P. S8 j! \! q7 n3 O/ ?4 W
  
• 
  
  这里是  Process-4
  
  
• 
  5 g3 r5 s' Y2 L, J6 G7 Y
  . U* a( ?" k8 k# Q* ?2 Z7 V$ b模块名称: __mp_main__
  + B9 a3 B" H2 x+ Y% Z
  7 z+ {! l7 U5 l# f5 X3 E
• 
  2 S1 ]: A; V3 W  @. I8 B% p+ R5 a$ K
  / z$ u/ Y: n+ g* ^' X( O) J: `父进程 id: 880
  
  
• 
  . R% x! \' o1 Y3 W
  - F! D3 O1 k3 Z3 \" x当前子进程 id: 5176
  
  $ `) G& b& N: V. f- I
• 
  
  --------------
  . m+ M  @- y1 h
  
• 
  ' j/ s% K, G! N6 b  q, J6 F* f
  . B8 U6 O7 m  u9 t7 m8 b这里是  Process-1
  / I% p5 ?+ O# z! F# b
  ' m8 }  b( z, q3 K
• 
  5 j  T1 N9 b2 e! W
  模块名称: __mp_main__
  
  
• 
  
  父进程 id: 880
  
  
• 
  
  当前子进程 id: 5380
  ! r& }$ @7 t2 C/ q; V, n: u
  * c  L  Q! D: j- v$ o: g7 j1 {+ _3 q
• 
  ( u0 g% j2 F$ R7 D3 i
  --------------
  / V) L# {2 X) O# F; M
  / o! Z# s: F- k- X* T9 a
• 
  % F5 O6 T" g) L. K3 C) c" _
  2 d# x/ d' Q7 C* ^' R* d6 q这里是  Process-5
  
  
• 
  , a+ }" I3 h8 Z( [3 E* G, B
  * U7 `; l+ i0 c3 I8 [4 V% E% N模块名称: __mp_main__
  
  
• 
  / B( x; j, U; g# l5 l$ r$ z
  父进程 id: 880
  ( D7 i3 V3 ~" }6 }  A0 k
  
• 
  
  4 h  H# s: d5 F4 a* C当前子进程 id: 3520
  
  5 k- l: M( j. z+ y6 X4 h/ X6 l
• 
  
  --------------
  $ X, n7 V+ }8 m6 `8 j
  
  ) K/ ^. K) ?/ H( H

1. 进程间的数据共享

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

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

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

• 
  
  from multiprocessing import Process
  
  
• 
  
  
  
  ! N! s2 S+ ^: X$ l3 N8 c) Z
• 
  
  + v9 W, m- I1 j+ q$ r$ xlis = []
  $ G- D: r3 i% }1 o9 G
  
• 
  
  5 V+ K" S8 P$ d/ X" F1 o) q- l
  
  5 j$ z/ E1 [8 l8 A
• 
  , ^' N; _; [8 s. O3 R
  0 t3 E0 R: c; x/ w( U- udef foo(i):
  
  - X6 N/ {* D0 e* y3 F
• 
  
      lis.append(i)
  
  
• 
  
      print("This is Process ", i," and lis is ", lis, " and lis.address is  ", id(lis))
  
  # L9 b* b/ W9 l9 W! R
• 
  
  
  7 ?% g7 s6 o: V6 j% ]% o
  ( W( @6 Q- `! l4 X: O
• 
  5 E5 @+ _9 w, X$ p: G3 E
  0 B; e7 l, [) G1 jif __name__ == '__main__':
  ) [6 n4 I2 ]/ K  Y
  
• 
  # y+ x7 p9 |# g$ q$ ]& i
      for i in range(5):
  
  5 T0 D% n6 J1 W/ r
• 
  
          p = Process(target=foo, args=(i,))
  6 {9 K$ a0 D: a% t- `6 J- Q7 {
  , R- ], |6 i' Q  Y+ W
• 
  2 G9 m  s( o; p
          p.start()
  
  ' m& W* n# I. z" ~4 O, }4 T
• 
  5 ~  J* @1 {0 Y/ v
      print("The end of list_1:", lis)
  ' ]" ]) t- d. s' p
  
  - w- M) ^% b" g6 B

运行结果：

• 
  
  The end of list_1: []
  5 b9 @( P9 P! v  v3 n6 W% W) G
  1 u) C, C: k* N' Z
• 
  
  This is Process  2  and lis is  [2]  and lis.address is   40356744
  
  % @- X! z* X/ a$ v
• 
  ) K/ o" X. G5 b) F/ i
  * H' B! n( M/ n/ c8 L7 u; p/ @This is Process  1  and lis is  [1]  and lis.address is   40291208
  6 }4 e0 s, B( {& i, T
  
• 
  
  3 m! P5 f: D6 P  i$ m; w7 zThis is Process  0  and lis is  [0]  and lis.address is   40291208
  4 ^- o- s; N* m' z1 @* x
  
• 
  & z, J& l) n: L
  % T* K4 ?. f# f  }$ g6 ]This is Process  3  and lis is  [3]  and lis.address is   40225672
  4 Z) ?! U: a! X3 n( G. u' H
  4 V. h! c, Y7 m, F6 W
• 
  
  . p& y9 U+ ]: V) d9 |, eThis is Process  4  and lis is  [4]  and lis.address is   40291208
  2 B' |0 ?: r; B6 u( y' _# C# O, S
    u7 }' l5 y! }( M) {
  1 W! U0 g. W0 L1 E) J5 H4 A

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

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

1.1 使用Array共享数据

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

• 
  % g0 Y& V) c2 A8 W
  - ]) G+ V& N$ p* M8 ]'c': ctypes.c_char, 'u': ctypes.c_wchar,
  # u+ V( n. ?6 ]. @0 ?4 Z9 y
  
• 
  
  'b': ctypes.c_byte, 'B': ctypes.c_ubyte,
  : |7 n  D+ H! }' d% T
  
• 
  
    ]3 v# ?+ P& b* i  ^'h': ctypes.c_short, 'H': ctypes.c_ushort,
  4 s% Z7 u% M) h3 _
  
• 
  2 k# Z- M9 \0 |' H1 l8 }
  2 D$ d; K# r# u8 y8 [* V7 w8 I0 }'i': ctypes.c_int, 'I': ctypes.c_uint,
  3 I* J: z3 `/ p, Z% F' P0 U
  
• 
  8 k( D, a5 d+ {! e( b# @! V
  1 u* a6 a( K, t- y( }% c" j: d'l': ctypes.c_long, 'L': ctypes.c_ulong,
  3 q7 u/ M, g, x
  % m. |$ n1 ^4 f8 B8 D8 K+ |
• 
  ; E/ F" V& J( r$ i
  'f': ctypes.c_float, 'd': ctypes.c_double
  
  1 l! z  m! c  |! l; Y
  

看下面的例子：

• 
  
  0 x! C/ l% z9 g2 Q5 B& {from multiprocessing import Process
  - J# C/ }) B( i  D, ~% Y
  ' ^" X0 R8 p% O: V& O' X0 J3 F
• 
  
  ) L' H$ N1 ^3 p7 l8 Q2 S; q' Efrom multiprocessing import Array
  
  
• 
  
  , s+ O2 J3 [4 t
  
  
• 
  - g3 [* ]: ?& [( A  y, H
  9 F( o; A- u; M& Qdef func(i,temp):
  
  / o# i3 Z% c1 b9 u; Q- [
• 
  2 i- Q! `! x( k! g
      temp[0] += 100
  
  3 v4 K) i8 r8 z7 g
• 
  
  ! u+ w) b9 [3 r* o# q6 P    print("进程%s " % i, ' 修改数组第一个元素后----->', temp[0])
  
    u6 c" d7 Z! W  X/ \; h6 h
• 
  
  
  7 O& g3 t: d7 a
  8 O1 _3 l2 E9 [% ~& J& L  B  ^
• 
  
  if __name__ == '__main__':
  
  
• 
  7 |' r) Y6 I; i9 L  Z$ k0 y3 m0 T
      temp = Array('i', [1, 2, 3, 4])
  
  
• 
  , g) L9 }$ n! P! p- X. s
      for i in range(10):
  9 b7 m& a" Q- `+ n$ ~. p+ U
  2 C  V( X; c8 S: @( _
• 
  ' f" h: G! q. X, b: J7 \3 L
  7 Q$ U" j6 K! v! |4 ~        p = Process(target=func, args=(i, temp))
  4 Y' C& T( k4 J
  
• 
  9 m/ e. ^' d7 i2 I7 L
  3 G0 E% j4 D, n! ]$ ~4 K- k0 \. }- {+ a        p.start()
  
  
  ! F: k: S" f0 s

运行结果：

• 
  5 E7 K6 d, I  o5 Y
  8 y4 R. u5 R$ K3 q  x- {8 B进程2   修改数组第一个元素后-----> 101
  
  4 L5 |* t/ B. r5 W2 f! e' k* B
• 
  
  进程4   修改数组第一个元素后-----> 201
  
  ' `6 A3 q8 K9 M' |) E% N, H, Y
• 
  0 S7 q) j" L+ e3 r: C, E
  进程5   修改数组第一个元素后-----> 301
  $ v, h  ~0 X! b, x. s) U- i
  - w- z" h7 e4 D- d2 H; z
• 
  
  2 \, N" V: e- T进程3   修改数组第一个元素后-----> 401
  
  0 d: S0 Y5 r! N' ]
• 
  
  0 j1 z4 `# o6 v! j5 E: }* N进程1   修改数组第一个元素后-----> 501
  2 q: D. X# @$ G' M4 ^& [
  + S# X8 q  U2 k6 O: A8 y
• 
  ( G: _" P' n  `- h2 s% K8 x
  9 C/ a0 d  F/ p0 f+ W+ d进程6   修改数组第一个元素后-----> 601
  ( x& M  l; m3 L* w' S
  ; z: H+ g: z" V5 s* e! l) p! H
• 
    t6 E- P& `* p; B4 R- q
  进程9   修改数组第一个元素后-----> 701
  6 b  B% G$ F* m9 d, }
  
• 
  
  进程8   修改数组第一个元素后-----> 801
  $ C. Q( t1 Y$ j
  
• 
  
  进程0   修改数组第一个元素后-----> 901
  
  . L5 K9 w6 A# R( V$ k
• 
    C1 x7 J% K- L9 c1 y2 X" ?
  进程7   修改数组第一个元素后-----> 1001
  9 d7 A# h( J# d6 Y
  % y9 {: e( J' S& a( O
  : ]* b! g8 k4 _  R  c

1.2 使用Manager共享数据

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

• 
  
  from multiprocessing import Process
  
  6 v" K' X9 K3 j7 y3 N+ n
• 
  
  from multiprocessing import Manager
  / N. ~4 \1 u1 k4 V( W7 V' l
  # y# y, [0 a8 ~( o
• 
  8 Q6 b# D5 q. G$ R
  $ a' r) ^! t8 q0 y; j3 o
  . p' a- Y( n' _5 H
  ; s* m: f* }! g% w9 d& `
• 
  
  def func(i, dic):
  
  - m3 n$ t. i$ E6 M  U* q
• 
  2 x% j9 T' |: D( Y/ u
      dic["num"] = 100+i
  4 ~- M: }7 D8 I
  6 `7 ~" w0 p: X8 E1 J$ X7 m2 S
• 
  
      print(dic.items())
  
  + k3 r% V, I! M4 O# d  h2 c' `
• 
  
  
  ( @+ I2 c& [6 z# U
  
• 
  
  : g$ Q- I8 {" y9 ]/ ]& Lif __name__ == '__main__':
  
  
• 
  2 J! `. X( K$ Y: ]% v8 d
      dic = Manager().dict()
  
  
• 
  
      for i in range(10):
  8 i0 k: L5 r) h  E) x4 q! m5 q# T2 Y
  - O( \8 B( I, p- e- G  z
• 
  5 R5 X$ c1 y" ^
          p = Process(target=func, args=(i, dic))
  9 N% D8 W6 q* a
  
• 
  
    @2 y2 F# `. S: \* T0 P4 K! H        p.start()
  1 O( m0 r7 Z5 Q' B$ V
  
• 
  + F, {$ Z, ~9 d2 G" T* R
          p.join()
  
  ' x: @8 R* n) [, D, n5 P! i0 ~
  $ }! T. o7 l, R9 u9 d

运行结果：

• 
  3 I, |- N! W# [  y8 s! n; e
  [('num', 100)]
  3 C$ z4 g4 m5 p/ a, s
  ( E' i! m$ ~+ y( g
• 
  3 r' v- X* ~, ^
  [('num', 101)]
  
  
• 
  3 Y( a* F/ Z+ M# t: f
  * D, }* u5 y% P) B( A  _[('num', 102)]
  ) g6 W  X' `% g( U
  4 }6 ^1 @/ f& j( p
• 
  
  [('num', 103)]
  8 ~$ m+ ?+ u/ z' S  F: a) q8 n& N* F
  ; t2 e' u, E7 b# [4 {& H. F
• 
  # |) h! X# W" B: W. k
  [('num', 104)]
  9 j% x' ?+ K3 d2 U& }( C8 \
  
• 
  1 R, P* A  f! T
  [('num', 105)]
  ( G0 K( i7 |% {6 r- k) {4 ~
  
• 
  
  , _' k: n/ o, w) b% f2 S9 d[('num', 106)]
  3 r: ]( [( i4 y8 Z- a0 T
  
• 
  4 q# A8 |( ?$ s. h
  ( w  O" |* R3 z[('num', 107)]
  + k, z  `8 X' g) u- w% D
  
• 
  8 d+ @* b: w8 E  `# p$ d: I
  [('num', 108)]
  $ P0 y. Y. b, N- ~
  
• 
  
  + n: d4 K- M+ S7 }7 l[('num', 109)]
  ' y/ Y3 d4 t3 `, ]
  
  ! r: Y/ f" x9 H% s( a7 a9 Y

1.3 使用queues的Queue类共享数据

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

• 
  - P; f' p3 X1 Z: G
  import multiprocessing
  5 G/ c1 g" `  t% O4 ?; A; V
  
• 
  $ Y7 L1 Y2 ?' U
  / \' a, h, E; Vfrom multiprocessing import Process
  
  4 F1 h; Z# C/ g) k8 x6 u
• 
  * k; A0 ?! a3 Q* B% X  W& S
  from multiprocessing import queues
  4 b* E/ z( E* G+ ^
  4 \, R  S: ^  I$ n) _: f) d
• 
  ! Q& ]7 b" _+ ?
  3 W$ f; I+ l. f  ^2 H) D7 h4 ?
  0 i8 }0 A  q1 K" s
  ; D6 V0 @& w! Q% i& ]0 Q. d' _
• 
  
  def func(i, q):
  0 A0 `- Z# v2 X* |
  4 Q6 J1 t; e& b
• 
  , b+ W) b- H: F6 m% z, x
      ret = q.get()
  
  
• 
  
      print("进程%s从队列里获取了一个%s，然后又向队列里放入了一个%s" % (i, ret, i))
  , T$ R8 i$ D! k& e8 V7 k/ i
  ! z4 @7 n2 Z8 q' U6 {4 t
• 
  
  8 O/ v* C4 Y8 [* O    q.put(i)
  
  
• 
  $ [8 Y" T1 ]2 U
  
  8 y7 p) t$ r- o
  : ?; n: {: O/ G: J+ f) d
• 
  
  0 h- _5 O/ O% c$ E* O; iif __name__ == "__main__":
  & M, ?  S4 ^7 N( x! Z9 l1 ~
  ( L' F+ x8 f* f+ y: s: b: Z% }& `
• 
  - E/ \! C' a% F( v
  . e0 j6 A6 s; ]4 \: ]& \    lis = queues.Queue(20, ctx=multiprocessing)
  8 R& R2 w2 ~. O
  
• 
  . J7 H' |5 Z# X8 T8 K* @6 X% L
      lis.put(0)
  4 P( n' [* U7 w5 j3 O4 [
  
• 
  
  $ D! U" y7 \7 T0 l3 N  d    for i in range(10):
  ! ]: x% y! z' U0 U$ ~; M6 `" n9 k
  2 i3 f: U4 b8 F* X1 W% N
• 
  
  : z4 i1 g5 Y% m0 }! K        p = Process(target=func, args=(i, lis,))
  0 H, u6 K0 w' R& w  r. ~2 l
  
• 
  
  , G+ Q/ ^: u) m; n/ y0 a        p.start()
  
  
  ; Q' {$ t& e' L* I+ n4 W

运行结果：

• 
  + B# ^; W2 S& {/ b; T* n" B
  进程1从队列里获取了一个0，然后又向队列里放入了一个1
    r0 T: b( v# T, q4 p9 H
  # Q( P- A+ S7 j. w0 c
• 
  - n! h4 c4 [( b: q& G! ^
  进程4从队列里获取了一个1，然后又向队列里放入了一个4
  
  : c3 J! s: O# n# N5 p+ e# R
• 
  6 s) K- i) X2 {& b4 k+ g, a" v# U
  " M8 G4 U3 F8 n  Z; n! i进程2从队列里获取了一个4，然后又向队列里放入了一个2
  ; h5 g& V3 o9 r8 ^! T+ _$ _' u; A
  + T( O8 }  w. g! c
• 
  / s8 `; F% O6 U; y% M
  进程6从队列里获取了一个2，然后又向队列里放入了一个6
  , a2 a8 e  f4 ?4 R* f$ L; y
  ; U! e& H! A% X, ?% i2 n
• 
  ; f# a' W+ J& T: \5 @1 F0 g
  进程0从队列里获取了一个6，然后又向队列里放入了一个0
  
  2 H5 q4 {5 c6 e+ C+ g1 R
• 
  $ X2 E8 g4 p/ z7 v6 F
  4 D% T6 S' {" ^进程5从队列里获取了一个0，然后又向队列里放入了一个5
  / w) l& R' q0 B9 n. |* v  u8 x& g
  
• 
  
  进程9从队列里获取了一个5，然后又向队列里放入了一个9
  
  0 V2 U0 D/ Z* p( |( T0 I- c  n
• 
  
  进程7从队列里获取了一个9，然后又向队列里放入了一个7
  1 L+ ^' n9 y3 P% ~9 K0 G5 v# G
  
• 
  5 B! a! B& k* f( x8 x: Q! e
  6 l+ w. \2 a& }9 O进程3从队列里获取了一个7，然后又向队列里放入了一个3
  6 m4 y* s; l% A
  * E2 ~0 d' `' |2 y9 O
• 
  8 E$ \7 E$ {9 q4 R5 y, E, x3 w
  , F2 _( a4 R& j  s进程8从队列里获取了一个3，然后又向队列里放入了一个8
  
  $ C3 \% [# H$ A) y- t
  6 J, i3 z2 N6 c* N' O. s

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

2. 进程锁

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

• 
  
  6 j6 f2 _) S# _# u  \from multiprocessing import Process
  6 S6 K* O9 \, w" G& b. |7 |
  
• 
  
  from multiprocessing import Array
  
  & g: Y: V# }. I% x" d2 s8 c
• 
  
  5 ^3 D3 }  L' ]/ Afrom multiprocessing import RLock, Lock, Event, Condition, Semaphore
  + B5 D+ q" b: u) p. ]5 z  }
  
• 
  
  , \# K/ _4 N( U9 Timport time
  0 {8 e3 P5 D6 G" ~; E
  
• 
  , w+ w0 \5 ?6 v& s5 O: K1 q- U7 C% ]
  % d1 i: w! j6 ]% s/ ]
  ( H$ Y3 O# T, y  O
  
• 
  
  1 K( X$ f/ z* p  F* jdef func(i,lis,lc):
  0 X6 z) F3 Y8 F2 S5 X
  ; P, P* F( X6 a& {! ]1 ~
• 
  # ?( n0 M0 A" J" {/ U
  9 f5 [7 ?# v5 N8 R8 Q" g0 i: N2 {    lc.acquire()
  
  
• 
  
  ! `- A+ J" l) f3 `1 ~    lis[0] = lis[0] - 1
  : [9 B) ]6 s' ]$ U! \7 N' J
  ! _4 k5 ]$ G9 m; E
• 
  
      time.sleep(1)
  
  
• 
  $ U! c7 ?7 B* v+ H+ a6 P6 I
      print('say hi', lis[0])
  2 A3 O. I+ |6 e1 O9 U/ a/ i
  
• 
  % w2 a6 G( H7 R) x& c
  ; ]- y5 y; v% @3 t$ K4 S1 g    lc.release()
  7 z4 S' g0 X) |8 r" Q' r0 m& N
  6 }: ]' D$ N; u- E
• 
  
  " o2 ^: R6 q) s9 s. i- H
  
  
• 
    `% ?* Q; O2 h0 z) i
  if __name__ == "__main__":
  
  8 d# V* b- @  L- s
• 
  - n  @9 S/ K2 J6 N
  + x' T: Z/ I! c1 v4 ]+ v$ Y    array = Array('i', 1)
  & g) w* ]/ r, F# D. w
  $ _! g9 Y' ^  K( {
• 
  ) D0 V9 M) s+ D: b  r9 V5 }" `
  6 D, j: f+ N. d- w  P    array[0] = 10
  
  8 A% S" T* \# s
• 
  + x+ c4 u  n3 }) g: t& H
      lock = RLock()
  % V6 y& c( ?; p' Q
  ! v9 N$ U- }. ~; f
• 
  " @; }5 H6 f" K; }, @) J8 A* ~
      for i in range(10):
  & v2 B2 n6 `( K; V( v
  & q) S3 y+ t3 U4 S) s  h
• 
  ' A9 ?0 V% S! z+ x+ m% h
  1 y5 j( f  _( p' z        p = Process(target=func, args=(i, array, lock))
  
  
• 
  
  3 w. ~- W6 Q' @8 P  m( ]2 v6 w        p.start()
  
  
  

运行结果：

• 
  1 b5 w, o0 ], h
  , s/ F5 t* L, r$ z. g4 c  R: esay hi 9
  
  
• 
  ) Z& b, B! [0 [+ x
  say hi 8
  . Q5 g: h3 o6 _4 I; z. k; E& f/ y
  9 B/ y  C  c" h9 z. \
• 
  + y0 G5 c" g( U' Z
  say hi 7
  
  
• 
  0 r# E( G" G3 U  H% w
  5 r; j8 C7 Y& j$ g! isay hi 6
  
  & d. b* H% {3 q8 _1 A" d
• 
  
  ' o) q, i: Y. u; qsay hi 5
  2 G% W' w! H; \+ g% F! N
  ( D- H- e9 J" m1 I8 N. D
• 
  
  say hi 4
  7 G. b# F8 t' Z: m  H
  # b  ^4 n8 l% i
• 
  
  say hi 3
  
  , W( h1 _9 M  i5 c/ X! a
• 
  0 r! m. F. r' d4 J9 M, d
  8 T- R' o  V" U9 Osay hi 2
  
  
• 
  
  say hi 1
  
  $ _7 D; b7 W" E- E& R
• 
  
  say hi 0
  % V) e3 M: y$ @9 ]0 a* A
  
  2 A. x" B% r! `) A, Q

3. 进程池Pool类

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

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

进程池中常用的方法：

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

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

• terminate() 立刻关闭进程池

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

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

  
  

• 
  ( [+ ^; W& t4 z" V5 a
  from multiprocessing import Pool
  
    t! G1 k( R- S* E4 t
• 
  
  8 M1 T" W8 V4 `import time
  8 ~6 O( e8 w+ T6 a! n8 t9 S
  
• 
  
  4 V* `5 x+ C2 B$ B& B
  
  
• 
  
  def func(args):
  
  9 z* N! F9 L# J6 d/ `3 U- K5 p
• 
  $ l4 \: Z* Z1 u" P9 {) V) z5 n
      time.sleep(1)
  
  
• 
  
  9 A! A% u  L' e  P- S6 Y/ M1 x    print("正在执行进程 ", args)
  
  * G: a# D% T$ J2 v4 o
• 
  
  9 \- I! G& P% g, ]5 I; R' {
  
  
• 
  
  if __name__ == '__main__':
  ) y; d/ ]) [7 n1 {& r
    ~' V' O$ v7 u* {
• 
  $ `" R0 Y: H8 T
  
  * J2 z1 F5 r! }9 b" K% e
  
• 
  % q! _6 s- m6 y2 Y) s6 V7 c+ F$ a
      p = Pool(5)     # 创建一个包含5个进程的进程池
  
  & J6 t% w  H% t
• 
  
  . K8 w6 B8 U" _$ y! v( e9 r# g' [6 P
  9 i) j; [- a: I8 d! K; i) D: `
  ! y& b5 ^# L# w* u9 p& ~
• 
  ' ?2 U3 E. N" J3 U6 X2 u
  # n* o$ b% T/ v    for i in range(30):
  5 [  A  u0 }7 u! D
  
• 
  2 F. e' j3 f+ h# x- j7 d, A
  / ], a; Q( R4 q9 H% ^6 |; e* D        p.apply_async(func=func, args=(i,))
  & o2 S+ O# U( v
  5 z8 U3 K6 f7 L
• 
  & A$ q7 P- B$ r; v2 `6 z9 }
  
  ) q+ s! E# Q& C4 m: g# m% E; q
  
• 
  
  * z, z9 p/ d4 F  d% c' J: a    p.close()           # 等子进程执行完毕后关闭进程池
  " U; X7 \  [+ |& c8 x% s
  4 G0 U% \; |0 d2 v& \; r9 c
• 
  
  $ g3 p+ i# o1 r4 G$ l- D    # time.sleep(2)
  
  8 z3 ]- e# K; l( O6 o2 \# Z( c
• 
  * i, C, X6 b% [" M9 u! [
      # p.terminate()     # 立刻关闭进程池
  * e" A; z' B. H/ p
  
• 
  
      p.join()
  
  
  & z6 d' X$ L/ }! [% {  }* ?. t

请继续关注我