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

杨利霞

爬虫（七十）多进程multiprocess（六十一）
# w5 ?& v0 z1 E. X4 X

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

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

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

• 
  
  import os
  1 J# z  Q7 B& |/ \4 E# ?
  
• 
  
  import multiprocessing
  3 A# f0 u3 W# E3 b. s3 Z" m0 J
  
• 
  ! \6 T% E* f- r6 j' C; c; y
  - H& E+ e* Q2 s! C! |
    j# M( j" F  ]- M  U) ~6 ^2 q
  
• 
  
  def foo(i):
  
  
• 
  
  & P9 q( ?7 q7 O5 T9 \- u  S    # 同样的参数传递方法
  8 K5 q3 q' S- Q
  , m, f( G6 r; l8 g
• 
  
  9 i8 ?* z! p( t) E- ~' b4 k    print("这里是 ", multiprocessing.current_process().name)
  
  3 _6 g: s9 y  R8 |
• 
  
      print('模块名称:', __name__)
  : H/ O3 o; w7 [. u: M( Q
  7 z, N; H; f) z* W
• 
  . E1 L7 [9 g* w" M: v
  1 S# _! a) X  F3 F$ X: `' p3 K  x    print('父进程 id:', os.getppid())  # 获取父进程id
    \+ |2 m6 n. G( w
  ; V9 \$ i* f  j) Z& o
• 
  # F$ h* b- I- Q! N1 C
  7 V) c7 _" w3 M. t3 Z    print('当前子进程 id:', os.getpid())  # 获取自己的进程id
  
  ) D0 E$ }, ?; |& o* E$ P! V3 }0 l8 v
• 
  
      print('------------------------')
  
  
• 
  
  
  
  ( e: D4 m; o7 Z; L
• 
  : F# E6 Y  Y* Z* C
  if __name__ == '__main__':
  " x( |6 S, Z: o
  4 k$ S( v' g& X, a/ j4 e
• 
  
  ! n. R& ]" W) w7 s7 L
  
  
• 
  
  6 Z+ G2 x) |' D3 l    for i in range(5):
  
  
• 
  
          p = multiprocessing.Process(target=foo, args=(i,))
  
  + A1 y- i2 Z' q! r, V; I( n! b& X
• 
  3 n" c2 Z" e3 ]0 r, j/ B
  ( n' S7 o, _, F+ ?7 ?        p.start()
  
  
  

运行结果：

• 
  % h8 D- ]) B5 j5 A
  , A! U# G9 l8 n! h* ]这里是  Process-2
  
  
• 
  
  模块名称: __mp_main__
  
  
• 
  
  父进程 id: 880
  
  
• 
  " Q9 j0 c. U$ f
  / D9 z9 H; N" |0 u% n当前子进程 id: 5260
  
  1 n7 _+ f( y" R$ y5 W: L
• 
  
  ) o4 c2 |  ]8 U( N1 s--------------
  1 R6 g& [* r; n( e/ t
  
• 
  / [4 b; y" E/ G' n: \6 t$ h/ `& x
  这里是  Process-3
  
  
• 
  
  模块名称: __mp_main__
  
  
• 
  $ V7 z/ u5 n9 n, c, ~! U+ M
  父进程 id: 880
  ; @+ J2 \* D7 _; Y  v, `8 l5 u
  $ o( e3 Y; h7 [9 D/ V; n' d* b
• 
  
  当前子进程 id: 4912
  : O; ~' R8 e! z, a/ ]
  
• 
  
  5 S6 s6 Q: F, u  {3 P--------------
  
  
• 
  
  / q% w3 [) V. d' q+ E( \9 O" j这里是  Process-4
  
  4 j+ C3 J6 c. n9 _& ~; |6 a
• 
  
  0 k1 A  }* u) V8 Q0 c0 M& N模块名称: __mp_main__
  
  ( Z" ]& [9 K- e+ V8 g+ d! X4 N
• 
  - z7 L' D4 o$ y) _
  父进程 id: 880
  ! X) |' E9 M7 s3 P  N" m, s$ p
  
• 
  7 z2 v* R$ M7 p6 t$ O
  % ~% c+ H8 a# \: r, c( s& u5 V当前子进程 id: 5176
  
  
• 
  
  --------------
  
  
• 
  
  : @( u5 A% \/ v) O, ]这里是  Process-1
  ; K' L. M! Z, S4 c- r4 y
  
• 
  
  & N( D3 u6 n' [模块名称: __mp_main__
  
  4 ?. u) L- h  O* l! m6 z0 a
• 
  $ t( G7 b3 L# h6 [$ W7 K, A
  父进程 id: 880
  
  . S* P' J6 V5 ^/ b# X# X
• 
  
  当前子进程 id: 5380
  / n8 W, E; s) I7 w3 k( R, ~5 T7 _
  5 p6 K* Y% X- d# z
• 
  - Q& q' f1 o* D3 B5 c$ k
  --------------
  
  
• 
  
  这里是  Process-5
  
  
• 
  4 l5 h' a$ i3 c8 ~6 h2 V- F8 |2 [6 A
  5 i6 {4 Q( x: _& |. w0 P模块名称: __mp_main__
  5 F- r2 ~. D: B+ I; A9 W; b
  
• 
  " M! G* q4 H! ^9 V/ `. f& Q
  父进程 id: 880
  
  
• 
  , N8 o( K3 {7 w& }- C! h& U+ m( L. D7 s
  * r% A3 X7 C. e+ `9 `当前子进程 id: 3520
    y5 G7 a* e% b, d( b
  
• 
  
  1 D+ `- z2 D# \+ ?6 s  F# M) W--------------
  
  
  

1. 进程间的数据共享

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

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

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

• 
  
  . D: P7 J. H8 n+ L. C. q, v9 \from multiprocessing import Process
  / |5 r+ i& ]0 A$ H
  
• 
  + N$ T+ H$ t9 F8 W0 |2 Z: R+ O
  0 W+ p9 e9 ~6 k+ J
  
  " X" t1 S* B. C/ B0 S; c
• 
  ; ~' o8 W5 J8 s0 N$ m# Q: B
  * W9 x% p6 U' c" d1 K, rlis = []
  
  % _+ ~3 P! A9 F' f
• 
  6 A9 K, D, U, e6 ~) V- g" r2 j
  # M; c8 }( x$ H
  
  $ _' k' V7 O6 E  D9 ^! V
• 
  9 h6 E( K- G2 ^2 R! F
  6 T* D, B3 l  {+ f" `3 T7 R, B0 xdef foo(i):
  / T5 m0 j2 G; C; J) P+ y# g' m
  
• 
  
      lis.append(i)
  5 B2 z1 \5 u4 z1 ]
  
• 
  
      print("This is Process ", i," and lis is ", lis, " and lis.address is  ", id(lis))
  ( P3 [' f/ M8 B" l1 o0 g  O
  
• 
  ( A/ W; H0 ]0 a: E/ }9 e
  
  / @4 h" D  c) L  R9 N9 ^* U
  ; |, X# L, v7 D- R
• 
  : ]  Q. j' U" {( ~8 g7 z0 ^
  if __name__ == '__main__':
  & Q% w$ V8 g, H
  
• 
  
      for i in range(5):
  # B( Q- |7 X- S3 s0 C
  # p+ j0 V3 O* W: g' u6 ]
• 
  8 B) y, m; A2 S
          p = Process(target=foo, args=(i,))
  - b8 ]# I4 M: F* G( Q7 I" {( d+ i
  ! f0 ^! ?0 r: K5 _4 D$ a& A
• 
  
          p.start()
  
  
• 
  5 Z0 Q& H' M4 u5 J4 \0 b
      print("The end of list_1:", lis)
  
  
  

运行结果：

• 
  
  The end of list_1: []
  & y; P. H; K$ l% y) F# G
  
• 
  
  This is Process  2  and lis is  [2]  and lis.address is   40356744
  1 ^" j  E1 B4 {; y# y
  ! h- t2 W6 X, @& u7 E# l' `; X
• 
  & c. h# H" u0 W; {$ @4 g  L! O
    }' b) T5 x  a+ r2 ]  N% h& `This is Process  1  and lis is  [1]  and lis.address is   40291208
  
  # x, t9 J% R8 H, @) a1 O3 |
• 
  
  This is Process  0  and lis is  [0]  and lis.address is   40291208
  4 H( Y! e& V7 x+ [
  ' S: S3 ]  S) G* w; |# I
• 
  
  - {3 D$ D& y% p) V& K0 `This is Process  3  and lis is  [3]  and lis.address is   40225672
  
  / ]4 Q$ @/ r" k# }0 Q# ~
• 
  
  ( b* r+ D! [8 {) L9 L$ P5 B9 W  C. [; x5 kThis is Process  4  and lis is  [4]  and lis.address is   40291208
  . Y5 D" A, e8 k2 t& K0 e
  
  # H+ G( ~( L8 g6 q& D$ C

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

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

1.1 使用Array共享数据

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

• 
  
  'c': ctypes.c_char, 'u': ctypes.c_wchar,
  ! _0 |3 U! m* B4 }9 U) e9 P# [
  
• 
  
  6 o, H/ [8 w+ X3 M- z9 Q7 l'b': ctypes.c_byte, 'B': ctypes.c_ubyte,
    X( r+ x0 s* {+ ~
  % w9 ^2 `5 q0 R$ j
• 
  
  ( Z  v/ ]" O# J3 `0 g+ P8 Z2 L'h': ctypes.c_short, 'H': ctypes.c_ushort,
  
  
• 
  1 @+ {  @: T$ A5 z: {
  1 L3 E9 Q! Q/ J' ]- `/ T2 d- x'i': ctypes.c_int, 'I': ctypes.c_uint,
  
  
• 
  4 A; k8 Y) D3 f% e! n( ?
  'l': ctypes.c_long, 'L': ctypes.c_ulong,
  0 B! z9 B: T5 [
  , S$ s; S9 s4 R4 \3 c8 p; w! ^
• 
  . t' \3 k5 i% Z- p. j% V
  'f': ctypes.c_float, 'd': ctypes.c_double
  . A$ H" R4 g4 L  d1 r$ _
  
  1 R7 ~5 m1 g  I$ Q. Y" E& U$ m- p

看下面的例子：

• 
    Q0 I' ^5 E' j1 s0 i$ L$ @
  from multiprocessing import Process
  
  
• 
  % g5 q3 F1 C* l0 \+ Z, w
  from multiprocessing import Array
  
  
• 
  ! @. E0 l6 U' X3 S5 ^8 _
  
  / A" `7 l& Q+ T, u; A, `3 e+ t
    j+ L3 d  ~% a  e8 v6 L  P
• 
  
  $ o  E" P1 z, t# H* ?- u1 Mdef func(i,temp):
  
  7 B+ u9 q+ T$ ~/ _: r. X
• 
  , l9 u) S& D' G2 H+ A; `% h
      temp[0] += 100
  & Z0 B7 g5 L; [9 L8 M
  
• 
  
  , P2 H+ l5 v# \+ K    print("进程%s " % i, ' 修改数组第一个元素后----->', temp[0])
  " n" L% C5 |# v  m  |3 e1 ~3 e
  
• 
  . V& L. c* c& X4 R1 R: P
  
  6 T3 D, P  O0 h# E
  - D6 O) e0 H, v
• 
  
  if __name__ == '__main__':
  
  
• 
  
      temp = Array('i', [1, 2, 3, 4])
  
  
• 
  
  8 P* K( s- h6 D1 r    for i in range(10):
  4 |. x! a1 V/ ]* {  G4 g
  , a1 R* s5 y% V4 m' _( I
• 
  
          p = Process(target=func, args=(i, temp))
  
  6 T; J7 g+ \* f
• 
  
          p.start()
  
  " _' v" y* B  ]/ O. ^$ c
  7 G  z: N% _7 W

运行结果：

• 
  8 w4 Q" q* K4 T
  进程2   修改数组第一个元素后-----> 101
  
  
• 
  
  9 ^$ D+ D8 N7 Y  a1 @# }5 M5 C& i. D进程4   修改数组第一个元素后-----> 201
  
  2 @: C; H$ ^8 ]' q2 x! G
• 
  + I  J* `) h; O9 w% z% [8 i
  3 x- E" [, r! M- E# L/ q进程5   修改数组第一个元素后-----> 301
  
  ) V- \, G+ b! F/ ~/ n' Q
• 
  
  / h7 N6 z" `+ A: P进程3   修改数组第一个元素后-----> 401
  ' g4 z: ?7 D' K/ J) u& Y
  + y% j, B( J# M! A) x* e, s  ?
• 
  
  ( L2 E4 a6 a! A* H1 R% R6 n进程1   修改数组第一个元素后-----> 501
  
  
• 
  
  进程6   修改数组第一个元素后-----> 601
    P' O: R, N/ f& Q/ r& L
  / d2 T' r' n6 D0 ~* z% @! X  Z
• 
  # Y; L0 L7 t1 r8 H1 w6 i
  ' y- D# }" d0 o0 Y进程9   修改数组第一个元素后-----> 701
  1 t' e5 {- p# }0 }6 o8 Y! y
  , ]2 c/ s- y5 f2 n( }4 ~1 D
• 
  
  ' ]7 o  ^3 N; m6 k; X/ {进程8   修改数组第一个元素后-----> 801
  
  0 S8 Z4 S9 z. N4 R
• 
  # u1 Z7 a: {' e
  进程0   修改数组第一个元素后-----> 901
  
    _1 ?/ I3 F. p1 P
• 
  ( M. z% O8 i6 a8 Z$ R+ ~
  0 i# N( V8 o9 G0 y% f2 g6 Y进程7   修改数组第一个元素后-----> 1001
  
  
  

1.2 使用Manager共享数据

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

• 
  + r) P& I0 [: c$ x2 W# D
  from multiprocessing import Process
  - @! x  _3 {: h" {; s6 q8 X
  
• 
  
  from multiprocessing import Manager
  
  
• 
  
  
  " s& {" j3 |* G4 D, {
  1 L& n! }) r5 l
• 
  4 F, O9 ?7 h% x. M
  5 g! H5 w, A) L; O1 G  a7 L- N1 tdef func(i, dic):
  
  6 r. W' j. k5 k  [! F: L2 @5 `
• 
  
      dic["num"] = 100+i
  
  ( p% }* X3 V2 J
• 
  2 c: k5 y; \4 T0 C& k
      print(dic.items())
  
  ; L: v' s: {+ Q, A- x
• 
  . h+ _, t3 R5 m: J% o
  
  5 ]: J; ]% u& k- |( M$ D* S. e
  
• 
  
  6 s$ D. V2 c% f4 T$ l5 Rif __name__ == '__main__':
  & S7 D( _! M" f$ M8 B2 H- s% B
  / `% _1 ?. E& a! J$ ~
• 
  1 v( A7 S" ]5 G
  6 }0 }# h' }9 z( J$ q    dic = Manager().dict()
  
  
• 
  
  & H) r$ n9 O  e% y    for i in range(10):
  
  
• 
  
          p = Process(target=func, args=(i, dic))
  
  ) R( `4 U) s1 l  t
• 
  
          p.start()
  
  
• 
  
  ( K5 D( X5 H0 g. A" H- q        p.join()
  & j. d& j  I$ ^; F% C0 \1 o
  8 X  I2 Q; i: w  f$ ?' M) l
  

运行结果：

• 
    x0 G, L1 p/ o3 O4 y! X
  [('num', 100)]
  
  % e$ v, w* ~! q1 a9 M
• 
  + M1 W( N; r9 L! I0 \4 `
  [('num', 101)]
  
  + ?& T, R5 g" Y: Z; k/ p
• 
  
  4 ^0 [5 K2 ]* i1 ?[('num', 102)]
  * o* ]% Y4 L# z7 C( ?
  & G+ z+ P/ n0 Y0 J; ]- u
• 
  
  . g, v+ [& ~  H; L+ a9 `8 d1 |[('num', 103)]
  
  ( v- u9 \5 y7 Y' j5 Z/ E6 l
• 
  
  # H: y. g( J+ n, z* C; D8 t[('num', 104)]
  1 ?" d! e6 x  [% H
  
• 
  
  [('num', 105)]
  
  / i4 y% U7 E1 r5 v2 P9 E
• 
  ( _1 E. i( Q: ^) }" Y
  [('num', 106)]
  2 e% c7 r' [8 l  `# d
  
• 
  ( _( F2 N# `# R/ I
  , R* y' W) E1 g1 b2 N[('num', 107)]
  8 r: [/ z) ^/ }4 ^4 X+ ~
  4 d% {) d/ y' I- ^" j2 H' M2 K
• 
  # M4 j5 D- p: w6 ~4 X' i
  [('num', 108)]
  
  
• 
  
  / P6 _2 ^/ F9 k0 d% B3 E; [: q[('num', 109)]
  
  
  6 b# b1 g/ y- Z9 |5 T) ]

1.3 使用queues的Queue类共享数据

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

• 
  ! _" V1 o. d9 c) {* \' `" Y4 ?
  import multiprocessing
  
    h. S: T, N9 s
• 
  $ I5 j- P# O4 n3 f
  ; i: _+ p7 Z) y" j0 S# x- ^from multiprocessing import Process
  
  
• 
  
  from multiprocessing import queues
  0 ]. j: j% H+ y- ^% O7 E; B2 s
  , Y9 {" X) a) L  p+ i: R
• 
  
  : f, c$ }- F/ c' @  w4 f$ [
  
  : q$ b( X: O& ~7 T
• 
  % R3 ^7 o- [+ P
  def func(i, q):
  + F5 d! D: @# z- M- I
  
• 
  
  0 r( r2 {' M- l    ret = q.get()
  / R# w7 l/ R! e  I
  ) O4 o) e. Z# |9 C( U; c4 f
• 
  
      print("进程%s从队列里获取了一个%s，然后又向队列里放入了一个%s" % (i, ret, i))
  
  ( x2 S! \& {* y2 C; d
• 
  
  : S9 `3 p+ @$ i% O' A: a: d    q.put(i)
  
  
• 
  
  
  0 g$ C6 Q. n! Y8 }) I& R/ n& C. @) f2 S
  ( N3 a5 o) @% O/ H9 k! h/ ?
• 
  3 ~6 ]+ a! ?3 P3 c8 v) B
  8 ?. k! c7 P% @4 ~6 A2 a5 sif __name__ == "__main__":
  ( A! w( S: r0 p4 M
  
• 
  
  8 N' ]+ v' \) x    lis = queues.Queue(20, ctx=multiprocessing)
  ( Y. x; R' p$ U2 W
  
• 
  8 l+ `% K& k; r. l1 {
  % i* V! v8 u+ H3 D    lis.put(0)
  : V3 P/ r' X# o) y% X
  
• 
  ; S4 ]1 s8 r3 r: ~6 Z; l7 W: w1 x
  , {& i' I! V* q    for i in range(10):
  
  $ G6 F( \# D& v9 n1 E1 m- g
• 
  : e2 L1 P( N' J- p( ]3 I* q& g
    K. G' t4 d, I: y8 T. M: ]        p = Process(target=func, args=(i, lis,))
  
  : W7 ^3 m. E1 B& R9 q5 R
• 
  
  ) k) `4 q' E% A7 ^        p.start()
  0 A7 M: s/ g8 S7 ?) G
  
  

运行结果：

• 
  
  进程1从队列里获取了一个0，然后又向队列里放入了一个1
  
  / S( Q: h5 K, V- H1 Q/ d/ R4 P+ h
• 
  
  3 s0 {7 B- Q& Q5 ~2 O进程4从队列里获取了一个1，然后又向队列里放入了一个4
  5 M4 H& m6 e7 Y8 T2 O7 V6 f: C* Q
  
• 
  9 Y) t( Z4 |+ h6 z/ G
  进程2从队列里获取了一个4，然后又向队列里放入了一个2
  
    b, @. m5 S! {+ r  R/ W
• 
  . l7 L3 f# R- b
  . T: \/ i( M- m  i2 C8 T8 r进程6从队列里获取了一个2，然后又向队列里放入了一个6
  
  0 l" S4 O1 q3 r7 z6 c2 a: W( u
• 
  
  进程0从队列里获取了一个6，然后又向队列里放入了一个0
  
  
• 
  - g% K0 t4 E! ]- B2 W
  ! s& |2 ~9 h! w4 V0 V6 t  I进程5从队列里获取了一个0，然后又向队列里放入了一个5
  * A. n. x$ }5 E- i1 }: h
  
• 
  
  进程9从队列里获取了一个5，然后又向队列里放入了一个9
  
  ; {2 a2 [. ^! c" E# f1 N
• 
  
  进程7从队列里获取了一个9，然后又向队列里放入了一个7
  
  ) ~5 Y/ W6 V+ q. k6 X; m, T! B5 B
• 
  9 Q. x1 i1 P, O: o* d
  / T+ z3 O- @( ?1 t进程3从队列里获取了一个7，然后又向队列里放入了一个3
  6 A! k+ j0 L* B' m4 D' ]5 P, v
  3 B$ {" M: K+ w: }
• 
  
  3 K8 U; V. I+ T  b, w" E进程8从队列里获取了一个3，然后又向队列里放入了一个8
  
  
  

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

2. 进程锁

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

• 
  4 }9 i; X% Z$ p: m/ ^6 |" o
  * {5 I5 G4 X( A9 R* cfrom multiprocessing import Process
  
  ; C, d# f+ ~5 z
• 
  
  * x4 w5 N1 }/ F1 J4 S+ G4 _4 `from multiprocessing import Array
  # i+ ^" v* s$ w1 }
  3 c8 x2 R! E+ G8 f- Z/ ~
• 
  ; q/ Z, Y9 [, P8 f- m- D1 g7 X0 x
  1 e# }& m6 d" i- A' J& w; _from multiprocessing import RLock, Lock, Event, Condition, Semaphore
  
  2 w* w8 Z" E8 v2 F# t7 W
• 
  
  $ x5 u4 T9 z8 Uimport time
  0 j: p8 w* t+ H0 S7 O7 N
  
• 
  " i( Y2 o: h( q" e. i8 p
  
  
  
• 
  : _9 B1 `& }) O, n+ E  @
  def func(i,lis,lc):
  
  
• 
  
  . N2 A4 L) E, J% q/ R    lc.acquire()
  
  ( y. K, O  Q+ t# L% S- z
• 
  
      lis[0] = lis[0] - 1
  
  2 a$ J! N0 U, g. H, }
• 
    g9 M6 G6 b8 v* J' n* j2 v
  & \! i: i3 _2 P3 U7 z: ]    time.sleep(1)
  
  
• 
  
      print('say hi', lis[0])
  % R# ^  B7 `1 T& y7 G. H
  
• 
  
  7 n2 H3 O0 @# v: y$ H    lc.release()
  7 ~6 D7 C3 B1 U( e4 p6 a
  6 ^1 {2 ]7 L5 C" U1 `
• 
  
  " N5 b* p) v5 p3 c
  6 f8 ]+ o9 |* J
  
• 
  
  2 {0 q6 s) f. Kif __name__ == "__main__":
  
  1 N! n* s$ V0 V& ^6 M& i) c  p
• 
  - S: F( D* [3 \, a
  # t' a$ l9 j+ n4 w7 e8 W' b  `    array = Array('i', 1)
  6 I* h; t$ O6 i& X) ^4 S6 |
  
• 
  
  & ^: ^' a+ h8 x7 a    array[0] = 10
  
  
• 
  
  * B$ Z* l0 |* d8 E9 n+ C# }4 j    lock = RLock()
  
  - W; S" g# R* E$ i5 w
• 
  2 Z* f9 T2 G/ p% F- X: ^( U+ V4 f+ a
      for i in range(10):
  # \& f5 P. E* O. P% b
  
• 
  % K6 t. A: z/ e1 \. g* S
  ; I; b/ ^' [- \        p = Process(target=func, args=(i, array, lock))
  
  3 g8 E  g6 ~/ n3 X/ J$ A# A
• 
    v! d0 _2 C8 G' M" l5 Q. A) \
  * d7 [5 g0 H  v; i" o) H0 v0 Q$ k/ r  N        p.start()
  0 J$ W  X+ j2 E' K
  
  

运行结果：

• 
  
  say hi 9
  
  
• 
  
  + t& b1 \* O# R7 Q, B4 `& `say hi 8
  
  / V1 \! b) `) p) |/ s( I
• 
  $ E& T7 C$ F; q
  - n7 q- `8 Z! E7 ?0 Z5 ]$ dsay hi 7
  $ ^# ^7 w2 G' |. ~: B2 [  g5 ^
  2 q6 E, D6 l; x! D! b
• 
  : p2 V2 m, f" I% U. l0 o  X4 R8 J; W3 \
  say hi 6
  
  
• 
  
  8 B! w' O0 M8 Y1 Z9 Y& ^: zsay hi 5
  5 f0 I+ a! G! O) a6 S
  2 |. n9 O, i1 U6 r
• 
  2 c% v8 E3 h1 G4 x! K" m, Y% a
  3 z9 R7 Q' \1 t' H6 }" U$ }& g1 nsay hi 4
  7 T; M) ?  ?8 c# \9 D! g& [
  
• 
  3 u1 p8 [( C. r$ x/ x3 Z! |
  say hi 3
  
  
• 
  
  * W8 c* ~9 A6 Q, o0 Ysay hi 2
  ) u2 S/ B1 F: ^. m+ k
  
• 
  3 i; a) f6 t( s0 }
  / u; |6 R7 [9 z& vsay hi 1
  
    m1 b' t  R9 h' H) G, O; j
• 
  3 q2 r* K; o3 I
  say hi 0
  
  
  5 v+ K: [/ l: ?! V

3. 进程池Pool类

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

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

进程池中常用的方法：

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

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

• terminate() 立刻关闭进程池

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

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

  
  

• 
  0 i, b: l- j2 i' }% q
  from multiprocessing import Pool
  * P$ c' k9 S) r6 ~3 _" z3 k1 ^$ ?
  * f3 K. A4 j+ H, n
• 
  ) z$ X3 S/ M/ b1 `: U
  import time
  
  ) F9 s- ~% I2 w/ y% Q% d( L
• 
  
  6 f3 G7 t  b. H, `" g
    z1 c& ]9 _: w/ @" Z
  0 L7 w7 T. @. a1 B& q" }2 k
• 
  
  def func(args):
  
  
• 
  : S  y! d5 M# V& c5 R% I, ]
      time.sleep(1)
  
  
• 
  ' q" y6 Y7 V6 r9 e& m
  . i* l8 X- l1 H  J7 T1 P6 i    print("正在执行进程 ", args)
  
  : x8 l: }+ K% `' W
• 
  
  ) ]$ ]! |6 D: X8 Q& n8 e
  % A, k5 S4 |: e( v# _( `& ?
  
• 
  ' j# G6 v6 D8 t8 P6 J5 O
  $ B8 g7 @# x, Iif __name__ == '__main__':
  
  9 V, }5 u$ B5 r* T& c+ e2 W# N
• 
  
  7 s7 r' S2 ^7 D6 W  _5 @  e+ |
  
  6 O: E4 l3 b2 c0 ^. Z9 {
• 
  
      p = Pool(5)     # 创建一个包含5个进程的进程池
  + m. `  F5 s* W6 y# v1 o' @
  
• 
  
  
  ; ~2 ?4 H6 V1 |  i% @( z' W
  4 ~# @/ a+ H% E; E0 C1 j3 A2 x
• 
  + }7 D& M4 D- [/ P+ g
      for i in range(30):
  
  
• 
  
          p.apply_async(func=func, args=(i,))
  
  
• 
  4 }! X& V* [: K, U- N
  0 f6 [& c+ I1 }6 \
  # \$ j) g( d& _+ i+ _* V" F- q' ]' H2 T
  ) E* S8 Y# d& v; ^# C
• 
  - y0 D, W/ u" @% N" i3 K* R
  ( I% E. X' t  e1 y9 d4 t    p.close()           # 等子进程执行完毕后关闭进程池
  9 E7 s% N' [. a. ^1 k8 C( e9 ^
  & ~; ]1 j, T3 p7 i3 `
• 
  
      # time.sleep(2)
  1 U& I, {& r% W. n- |- u/ p
  
• 
  3 \8 J0 d0 a& a8 a# B) L
  * n3 z1 m$ x( ]0 R+ m3 v1 V/ P    # p.terminate()     # 立刻关闭进程池
  
  
• 
  $ |/ a. A3 |8 O4 r# N
  5 C( B( u3 l6 ?: q2 B, p6 Y8 ?    p.join()
  
  . \5 c5 m+ O7 o
  3 h& d+ _3 W- k1 n  h; Z2 T9 M

0 E3 S( f5 e1 A  g1 Z+ s

请继续关注我

1 `7 N3 G/ s& ]) F) L
- u6 S2 U4 k: ^8 K% A5 _) G* t