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

杨利霞

7 j) u" |# h2 K2 O7 v爬虫（七十）多进程multiprocess（六十一）
* s" P/ @1 S' o: y

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

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

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

• 
  
  import os
  2 L: ~- U) e! C3 F9 u$ v
  8 w9 T% m# a0 k6 `$ }1 M! V
• 
  : M# H0 }6 M0 k/ \5 z  _
  3 w% h1 h9 J/ W/ f8 j$ Rimport multiprocessing
  $ C3 v3 K8 |/ D7 {
  
• 
  , t/ i+ Y; G8 @- c& x/ L% @: v% T+ Z
  % Z$ w( B* N4 N! q. N
  
  
• 
  7 p: h0 b+ J4 x; m
  . c( L! }1 P% D8 Fdef foo(i):
  
  
• 
  
      # 同样的参数传递方法
  ( L# W( d+ T9 B
  
• 
  
      print("这里是 ", multiprocessing.current_process().name)
  
  " j# h& Z' {$ m( c
• 
  
      print('模块名称:', __name__)
  
  & f% u- o8 a- u+ H
• 
  
      print('父进程 id:', os.getppid())  # 获取父进程id
  
  ( Y, C' k% ]5 S- f; d  e
• 
  
      print('当前子进程 id:', os.getpid())  # 获取自己的进程id
  ) }6 ~) [0 Q0 \1 q( E' K
  
• 
  " @6 t# n. C5 Y4 ]4 K3 _
  ) D) X, e0 h8 U8 F. L' o$ {3 }    print('------------------------')
  9 C8 ~* l1 H) ?3 y0 H- @
  
• 
  + Q5 R$ Z$ R. o# [2 r4 R
  0 z% E8 @; e& o
  
  
• 
  
  if __name__ == '__main__':
  
  
• 
  ' B+ |  e, t% U/ b  M
  
  / K# |% G: t' z2 [5 E$ G0 `: i# M
  
• 
  
  : x# K& }- B( K6 p/ s8 S1 ^    for i in range(5):
  
  2 @/ m$ w# E  r0 O1 J0 F
• 
  
          p = multiprocessing.Process(target=foo, args=(i,))
  8 ]6 n8 o, k( t: l- z
  ; i9 T7 y9 K3 P9 F
• 
  : |4 [& Z) I+ B4 \* f
          p.start()
  
  ; ^- @! n. Z- U  ^5 [- m( d
  

运行结果：

• 
  7 V" t4 B2 x' j' y
  这里是  Process-2
  
  
• 
  
  ; Y2 k9 }) w* S  E模块名称: __mp_main__
  
  
• 
  8 n- ~( T- |" G# P0 h
  - N4 ~  j9 L6 n. z, `父进程 id: 880
  
  5 [4 P: e* r( i8 T+ d" }9 V1 `
• 
  7 }0 B4 v1 B# c6 Z1 U& o) ^* p( C/ G
  当前子进程 id: 5260
  
  
• 
  % I# G' W7 t; Z8 a* D! i
  --------------
  
  8 y; Z2 ]7 B  v- @- O
• 
  ' B5 C4 f2 V4 a+ _0 j. G
  这里是  Process-3
  ; n; ?- q0 F2 [6 j8 ~, x3 \
  
• 
  
  * v8 q- m3 w6 Q& n- [' Y4 J5 ^模块名称: __mp_main__
  ) u0 M' R1 @+ t
  
• 
  
  父进程 id: 880
  1 [6 ^$ S+ d) S( P9 Q% z3 n
  
• 
  
  当前子进程 id: 4912
  
  
• 
  
  8 S$ K7 a6 c, m( T4 n$ [5 O* W! Q! f--------------
  
  
• 
  / @8 |9 c  ^- P. B6 v
  这里是  Process-4
  $ D. ]4 x) E6 a) \/ `* y" G; q
  
• 
  " z; Q. y9 m4 [
  模块名称: __mp_main__
  # J0 O7 Q6 E1 t. D
  , Z% {- c" @9 ~* z$ U
• 
  5 c8 L9 T/ u4 s+ J- s+ w
  - Z" Z( n$ Y& Y8 v% Z. W- H* j$ G父进程 id: 880
  
  % d) K) g6 G4 U
• 
  + @) l+ B. o+ G5 G( n5 e! t1 ]: H, G
  . H1 x$ {# y+ D8 k) `. C当前子进程 id: 5176
  
  ) ^+ B6 \  V; P0 G3 z' v
• 
  % T& O4 J+ _8 D- s& \6 e6 Q
  , E3 f( G& W  H7 v7 d% l--------------
  " Y9 ]  w: |$ |% F# b3 }3 i
  
• 
  
  这里是  Process-1
  
  " V! v, V4 h9 N$ C- y1 b
• 
  & {* g) f) M6 C" u: ^0 D3 o  \4 |
  % B$ W; ~" a+ q1 F8 I0 e模块名称: __mp_main__
  # x  K" g! U: o* X+ ~
  
• 
  ' l, r/ y, u6 a4 M$ N' q& q
  父进程 id: 880
  ) y4 g0 D& h. b$ |: D3 D1 v! W! ?$ D
  
• 
  
  当前子进程 id: 5380
  7 s. Z. X' R! @: g7 Q; t
  3 g6 E/ w. y/ e. s; n8 e
• 
  
  2 ]6 Z8 i! A/ }6 h4 }--------------
  % e# H# r. r7 o& O0 N
  
• 
  
    a, ^9 W  u' |% {这里是  Process-5
  
  # X! r+ j+ i0 x3 K
• 
  
  模块名称: __mp_main__
  4 [: ^6 _! _) ?
  
• 
  
  / s: x: L; P8 i. T# ~* h% k9 i父进程 id: 880
  
  
• 
  1 I/ W( q. }7 o$ I0 r8 G# _
  3 A0 C. D- \8 b. h; o# s# K当前子进程 id: 3520
  
  0 t8 e" W& p: E9 p$ B2 T
• 
  
    N8 t; i& o- g  W--------------
  
  3 j8 Y1 l( L7 i/ a- k4 N: N& G
  

1. 进程间的数据共享

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

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

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

• 
  
  ; v, }) i) H) B+ e2 a5 t  @8 Ifrom multiprocessing import Process
  . o+ `5 z* ]) s
  
• 
  6 z$ s" y. Y( ~1 s8 Y- p) S& @
  : k# F4 U# i; g% B$ C8 T
  
  ! i& N% S; S1 k& g0 Z( \5 Z: P
• 
  ( n3 I" Z; f$ ~1 c  n
  8 V8 z& Y+ B+ e# f% Glis = []
  
  
• 
  % P- U+ G* g% [8 l
  
  ! l) M( w; _) K0 U8 }5 t9 p4 s
  . ?) g- Y# u6 a% l. E$ I8 D) H
• 
  : ]8 K+ Y- a. t& J! k
  def foo(i):
  % _' t6 Y# b  l4 d: I' A: G7 l
  : C. n1 H& D  ~9 \
• 
  $ R' Y9 n; w* [7 J- Z  s4 c2 |- d/ n8 `
      lis.append(i)
  ) }2 h) B; w6 J1 ~2 F+ N. q& z
  4 V5 M# _1 r  |" ^/ v
• 
  + r9 c- `) k& @
      print("This is Process ", i," and lis is ", lis, " and lis.address is  ", id(lis))
  8 C; A. v7 i5 i9 F8 I5 ?. V
  
• 
  9 f' v2 s' v$ O% t" C: {  c# @' g
  0 i8 b5 l+ t! e) r- |+ p. x
  ' v3 `* N+ V* y1 p2 k$ {) K& d9 i
  
• 
  
  9 X( R3 x2 A4 k! dif __name__ == '__main__':
  
  
• 
  
      for i in range(5):
  
  % S* M3 E2 C1 P4 n
• 
  
  3 A6 \  x9 R4 H; o/ \. N5 |' [        p = Process(target=foo, args=(i,))
  . R: B- c& ~( S" n) |' z0 N
  
• 
  
  3 M0 n( W. e3 X! L: W        p.start()
  ' g+ U- @" n. R! I
  ; b5 ?; S# E% a
• 
  
  2 @4 s2 Z1 l2 n! c0 C    print("The end of list_1:", lis)
  4 l+ N9 Y9 h( e! G
  9 X- f) b7 n% w  T: j; a
  - s' f0 }5 T* M2 V1 s

运行结果：

• 
  
  The end of list_1: []
  
  0 M5 t4 J8 W0 e4 }; d$ n# N! c
• 
  ) k* V/ r$ X3 `. _
  5 ~& k6 F; D4 X9 |This is Process  2  and lis is  [2]  and lis.address is   40356744
  6 `0 O1 ?" l* M$ l+ b6 h
  
• 
  
  This is Process  1  and lis is  [1]  and lis.address is   40291208
  : X; ~- f* t% A4 p
  
• 
  
  This is Process  0  and lis is  [0]  and lis.address is   40291208
  & w1 P) D: ]7 h7 n4 t2 w8 g' {4 Z% p
  
• 
  , f0 {/ C* S2 l; b( h
  . n: ]6 w3 M  O; a3 FThis is Process  3  and lis is  [3]  and lis.address is   40225672
  
  
• 
  
  This is Process  4  and lis is  [4]  and lis.address is   40291208
    T- l$ h( ^* j4 B  H# p3 l6 d! N
  * k! Y( R8 R+ [% C) u( U, {9 @
  # C5 w7 `/ F3 h% |) X/ x- f

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

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

1.1 使用Array共享数据

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

• 
  ; t: m: }& d+ X$ h
  & O7 F. t! l# O# a4 Y. X, f) m'c': ctypes.c_char, 'u': ctypes.c_wchar,
  ' T# _: v! p: u
  
• 
  & ~8 C; S6 U6 _# F' V
  - m7 b* T9 s7 j4 |  k" y'b': ctypes.c_byte, 'B': ctypes.c_ubyte,
  ' I9 p1 b- t6 I2 V$ O: G$ K
  4 f! ^7 \$ H& Q6 J
• 
  
  # S% C) L6 i  K+ ^5 p) V$ R'h': ctypes.c_short, 'H': ctypes.c_ushort,
  : A" s. J5 }. v0 \8 r
  1 s9 G$ l, c. x& Y+ @
• 
  
  'i': ctypes.c_int, 'I': ctypes.c_uint,
  
  
• 
  3 m  r: q2 e: _& P
  'l': ctypes.c_long, 'L': ctypes.c_ulong,
  
  0 q4 n+ I3 M! _7 W, y! ?& |
• 
  * d' g0 U0 E( K( e9 P& T2 Z/ A2 y
  'f': ctypes.c_float, 'd': ctypes.c_double
  : o  o# u7 {& {! {* k
  , m8 m" E( t) ?- i7 X; a6 c6 N5 |
  , Q+ i  V9 w! _; P

看下面的例子：

• 
  & d+ O( A% @7 Y: T
  from multiprocessing import Process
  
  
• 
  
  / b, Q/ g1 \3 _9 Ofrom multiprocessing import Array
  
  
• 
  
  ( R6 X, q4 w1 m7 a, t) v% k
  ' O. ?' a8 a, L! l! O1 o6 Z
  
• 
  * \1 J" K/ X9 t4 k8 t
  2 M4 N; O+ Q! o4 d" G. Mdef func(i,temp):
  2 t5 {6 r$ l* w" W
  ' v! u% p+ ^1 L- S, P
• 
  
  3 V/ H, m0 k) m; {    temp[0] += 100
  
  ( m* e" p3 G1 ^7 B' x
• 
  / k/ V2 [" H7 M6 ~5 M7 J& D: b- N
  ' i$ C/ r+ m1 K! `7 \    print("进程%s " % i, ' 修改数组第一个元素后----->', temp[0])
  
  - x# O" {4 K3 V8 x. D2 _
• 
  
  5 M& |! K5 u: M
  $ p' g% o. A: a* E
  6 S" t0 \/ W. m, M, x4 _
• 
  + B" G0 p7 ^* @6 H6 _0 T
  if __name__ == '__main__':
  6 j: `0 Y- P/ u  {( x
  - S; ^% ?+ T$ G0 u& Y
• 
  
  2 l9 B% J/ b. @8 M    temp = Array('i', [1, 2, 3, 4])
  $ W  v+ u7 A  ]0 s% h" i
  % `9 m5 `7 h5 Z1 m! `6 ~2 w
• 
  
      for i in range(10):
  
  ' B: Y7 v# I  M( U  x- V! h! R
• 
  
          p = Process(target=func, args=(i, temp))
  ) }" A8 m8 n. q) H# d" y- h0 M+ [
  
• 
  
          p.start()
  ; \9 b7 b9 y7 U) {/ `6 U# t- t
  
  9 ^2 Z# C: `3 K$ l- J9 ?" u' d

运行结果：

• 
  , r5 F8 Y- ?0 |. i& c; u
  3 t3 ]& e) X% k" e6 ^进程2   修改数组第一个元素后-----> 101
  
  1 f: ]4 @4 U/ k( _8 P1 q' C
• 
  1 B! h. S9 A/ P3 |. Z9 \' @+ _
  进程4   修改数组第一个元素后-----> 201
  
  
• 
  0 u" P; y8 ^3 ~6 S8 C, g+ C1 h5 p
  " y0 q% p8 |! G9 o: a' p进程5   修改数组第一个元素后-----> 301
  2 ?( p, }6 c& [& u
  # x5 C; v! R6 y
• 
  & k) o3 S/ q0 j, h
  进程3   修改数组第一个元素后-----> 401
  - V6 Q& v* l, h. P; e. ^0 O5 \
  # U+ n0 `- q! {" u+ [1 z
• 
  
  进程1   修改数组第一个元素后-----> 501
  ) ?7 t3 l. C' L5 I
    Z* @5 w9 V) S' H& }" D! E) A
• 
  
  进程6   修改数组第一个元素后-----> 601
  6 n# ]) T8 A7 R+ u; x& {  X
  
• 
  
  * ?8 k: C5 l& I) `7 {进程9   修改数组第一个元素后-----> 701
  
  6 D) S6 \! F0 ?( F* i4 ~
• 
  
  进程8   修改数组第一个元素后-----> 801
  
  9 F" [5 W2 h' t; ?% I2 Z3 X
• 
  
  进程0   修改数组第一个元素后-----> 901
  + H* S# j& [- n2 ]9 x5 l
  
• 
  
  进程7   修改数组第一个元素后-----> 1001
  % h. n' k. o: P4 I
  7 A! k$ h8 O! I9 Y+ B9 a7 {4 g3 ~% g
  

1.2 使用Manager共享数据

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

• 
  8 x# R% }7 n# T& I+ R
  from multiprocessing import Process
  
  
• 
  / Q4 y) J7 t0 ]: G% D. u
  , ?' }1 P: {+ ?" @, |" S. H* ~from multiprocessing import Manager
  4 L) k% b4 q; B
  ' c- d0 x, V. ~; L
• 
  
  
  , @/ d  R% }. @: b( k% B
  2 |: j- ]4 }) k8 b& D% }4 v
• 
  
  - V: X( i, P# [# N" Adef func(i, dic):
  * l2 j7 {" ?- H# N! K
  
• 
  
      dic["num"] = 100+i
  
  
• 
  
  6 W1 P! H. m) ]0 h' J, x    print(dic.items())
  
  3 q. [5 S3 v4 E6 u: o, A
• 
  8 v0 @/ k/ \8 E, o# f: X  R
  
  % [2 x, q) R- p. D
  ; R6 k; A7 X) E8 C& h+ B/ X
• 
  
  if __name__ == '__main__':
  7 r* p7 E1 b9 z, Y! a; A. [  B
  8 ~/ d* w* L* C$ `- z; [* F
• 
  
      dic = Manager().dict()
  
  
• 
  2 U/ e- Q" i( \" P, g
      for i in range(10):
  . r2 a8 V, o' E  `& U8 T
  7 }9 r8 I, b6 G- d: e8 l6 q7 M
• 
  
  & j" y* S2 x% l3 z" f        p = Process(target=func, args=(i, dic))
  
  
• 
  * h- o& P2 u$ ?( k
  ( D) Z" b6 x$ e- m5 n        p.start()
  & ?  f; b! J" d) R# P2 P8 F9 S( c
  7 R5 [, F/ A( k" t+ V; d
• 
  
          p.join()
  ! ~9 b+ g( _- R$ P# M! _
  
  

运行结果：

• 
  ( U- u/ ~# X: d2 }; L% {( u
  * C- d8 ]% b2 w* J[('num', 100)]
  4 K7 J& S7 t- L& y7 m
  1 H0 q% B% ?( @1 v' a
• 
  
  , X2 W- t# }; `" }[('num', 101)]
  9 L5 j6 ~( ^: W0 @5 p- I
  * `' i/ Y' e2 ~8 `6 c
• 
  
  ; P9 z' e- Q7 D4 }* \3 z[('num', 102)]
  9 m6 t. S3 u* e
  
• 
  # ~5 w6 C& U2 e; g9 ?
  [('num', 103)]
  - c- y- D  u( E- U+ @  V( `
  7 D- T+ ~% a% U
• 
  9 {7 [( O& E; q1 }; K9 S7 o; A: E
  2 ^. C7 M9 S0 R[('num', 104)]
  . h  s: `$ g, F7 B0 D
  
• 
  6 y* F( t1 h1 x! C
  [('num', 105)]
  
  
• 
  7 X2 Z, S8 a2 ?! ^: r0 i& {0 T
  [('num', 106)]
  
  
• 
  
  4 }5 w- G4 k8 y# [[('num', 107)]
  
  
• 
  
  7 }, V, V( N8 X[('num', 108)]
  
  ) P- }1 ]# v( ^+ M# ]$ L0 J/ ]
• 
  
  5 d" Q! Z& ^5 T[('num', 109)]
  ! G5 c5 y5 O4 g6 }* g5 [, Y
  
  0 \3 N% ]3 \8 P# U

1.3 使用queues的Queue类共享数据

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

• 
  6 k0 h& `8 D9 w& G: d
  import multiprocessing
  
  
• 
  
  from multiprocessing import Process
  
  
• 
  ; q0 Q# k) e' j* ?
  from multiprocessing import queues
  1 d2 w; M5 M! A0 ?
  
• 
  ' M$ X# {/ d: z- q
  8 q- w6 A4 y1 H
  
  ) t& C0 f, ]' S. A5 i
• 
  ) L; h' F1 s! T: f/ d2 r/ G
  : w( V+ D6 K" g! Cdef func(i, q):
  * q0 b/ n' V+ w4 o, O; a. |
  
• 
  
  # E1 o0 v' A5 [' q    ret = q.get()
  3 X) c* U9 F# m; Q
  
• 
  
      print("进程%s从队列里获取了一个%s，然后又向队列里放入了一个%s" % (i, ret, i))
  
  
• 
  + j" E4 e% \/ P# i( R1 c
  - W$ q/ o% c+ T    q.put(i)
  , ]: W" w: x- u) [
  ( V7 Y$ O( m8 s& p) Y
• 
  
  . V# d3 V2 k$ W1 {0 C
  
  : E) F' Y" g! S& T, _# y( A+ w9 T
• 
  * v+ b: }& @% ^; _/ k& \2 i" _
    G% \4 m8 p! z* eif __name__ == "__main__":
  
  % M: M0 l3 b* d% k  x3 d
• 
  
  5 J8 Y; k1 ~1 O" L. }    lis = queues.Queue(20, ctx=multiprocessing)
  
  
• 
  ( K; F# ^! T# t, f' z! D8 T
  0 [1 J% U5 s0 [    lis.put(0)
  
  
• 
  
      for i in range(10):
  2 s5 v0 U' F& S) M/ k6 w' K: @
  % L2 {5 {" S3 i# @$ p
• 
  ) H6 E7 B- l$ L/ B4 I
          p = Process(target=func, args=(i, lis,))
  
  
• 
  
  9 A! B3 `/ n. O3 t$ {- b        p.start()
  
  5 l. s1 Q" l* Q3 `' i% _) b& G
  # ~" m; l4 H3 j- Q. w

运行结果：

• 
  
  0 A2 j5 n2 U! X进程1从队列里获取了一个0，然后又向队列里放入了一个1
  
  " b$ f5 z- D5 g% p/ V( a" z; B6 P
• 
  9 P  z2 O8 \& i* K4 k: X, k
  进程4从队列里获取了一个1，然后又向队列里放入了一个4
  
  % r! o+ r1 @9 G) V* |8 c% k# M+ f
• 
  ( N/ l# {# \3 K0 S
  进程2从队列里获取了一个4，然后又向队列里放入了一个2
  0 N2 i* |1 t7 o4 P6 p* U8 H  W
  / j1 o5 Q/ C: F8 ?) A5 `
• 
  
  ( N* }; a* k9 x1 C2 D进程6从队列里获取了一个2，然后又向队列里放入了一个6
  
  7 a; R+ n' O& n1 M) t
• 
  9 B$ t0 q7 n4 [9 T
  进程0从队列里获取了一个6，然后又向队列里放入了一个0
  
  
• 
  
  ( k; J% w' R1 k0 F6 @5 h进程5从队列里获取了一个0，然后又向队列里放入了一个5
  # T# S/ t' C( l3 p. s
  * }% S+ ^/ k% T" F9 o
• 
  
  进程9从队列里获取了一个5，然后又向队列里放入了一个9
  
  
• 
  + C, T: N! F6 U$ C+ G" j! u! d
  : y( q* ]. t* a7 f8 m- Y进程7从队列里获取了一个9，然后又向队列里放入了一个7
  ; E8 g$ o6 M5 B% ~9 ^+ q/ i2 E( [& }
  . K9 ~( _, o6 A
• 
  5 Y4 R( v* |% m/ m$ W& f
  . `, ]' t; f5 y9 d; w% ?进程3从队列里获取了一个7，然后又向队列里放入了一个3
  
  
• 
  8 q. t( `" E% g6 h1 u! u
  + I2 h  c: @# ~: `* U进程8从队列里获取了一个3，然后又向队列里放入了一个8
  
  5 z: O. V% |* I4 {
  

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

2. 进程锁

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

• 
  & @# N& ~* u7 b8 X1 ~, I
  from multiprocessing import Process
  7 p1 p/ n1 V/ Z* q! s- Q5 b
  ! Q6 R  f1 }1 J4 L- M2 a
• 
  8 Y  ]/ U* N' [
  from multiprocessing import Array
  
  
• 
  $ W0 G2 [. b1 M  n9 B- T
  : w% @5 n/ h* K0 H7 E: ?from multiprocessing import RLock, Lock, Event, Condition, Semaphore
  " h+ N4 W, t8 ]$ I1 o9 X+ I; r
  % |7 }* ~* L, y- a; x' [" _2 d
• 
  
  import time
  
  
• 
  
  2 \- }! ~' P# T( \- E8 Z7 T4 a6 E
  
  
• 
  
  % c+ M! d: e; Y7 n2 ydef func(i,lis,lc):
  
  
• 
  
  0 G# M" ]& @, Z  G' F: a( l! C3 r6 |    lc.acquire()
  9 h4 H" b1 g& U
  
• 
  
      lis[0] = lis[0] - 1
  
  ) O8 [0 m; a  K3 W
• 
  
      time.sleep(1)
  # m8 o, @6 N' ~) p- O( }4 m
  
• 
  ( _. \0 w. O! h) h' w
  , n- F# l; c; O    print('say hi', lis[0])
  2 b! _& W+ f$ N+ p
  8 K0 K: _$ a( O5 _* p4 B3 {
• 
  
      lc.release()
  
  
• 
  2 x7 O" o1 d; g9 s: i
  1 P3 f1 Y% J( b3 U- \
  
  
• 
  
  + f" Z: Y) U. l5 ^7 N4 j$ yif __name__ == "__main__":
  
  
• 
  ) v6 l2 `7 k5 p7 C  q9 ?$ p
  # L& [( P! _- p& Q# S* ]* f    array = Array('i', 1)
    G  w) w; H3 C9 l$ g
  2 j# n6 N1 C7 H  @) j
• 
  : l- C- N3 ]' y& A) c
      array[0] = 10
  . a5 a7 S4 r; [! n0 f' f! g
  ' F- W+ i2 N; h; m4 R  D  M/ i
• 
  # i/ P# p7 \1 d9 j1 p- g
      lock = RLock()
  
  
• 
  % f2 D3 o+ O' H, l: z, `# ?$ @2 Y) O
  ( ^- m. x, W: p& l    for i in range(10):
  9 v4 F' M/ g* N+ o; f: \
  / e) d0 K; k) \! O( Z+ q1 M- m
• 
  ; j  G* W+ G  r* S9 \2 E. k6 `+ u
  9 f# n7 L& r& l) U        p = Process(target=func, args=(i, array, lock))
  2 K; o5 x. s% Q& R' `
  
• 
  
          p.start()
  
  ; c+ I4 ~/ w1 ^8 R' S( F
  

运行结果：

• 
  
  say hi 9
  4 b9 h. [% Q4 U, J6 Z
  8 r# a7 }9 ?, q+ H' S
• 
  
  say hi 8
  ' D$ `& g% @1 X8 X! N0 N
  * U' }/ m# _. ?  T* Y% V' ]) M
• 
  
  say hi 7
  
  
• 
  
  say hi 6
    |- y: V2 \) t$ f
  
• 
  
  % }$ l! B6 G, @5 o- ?  Y& |say hi 5
  
  8 Q8 i3 S1 g+ s# x, {
• 
  4 b- I9 I% W  K) ]
  say hi 4
  # R% X  Q) d0 x# j5 ?) f5 j
  # \, \+ \8 \+ V- F$ S/ D9 p6 v
• 
  
  + U1 u! y* z, o# p( s; esay hi 3
  0 @' W+ g  C; D% f0 r1 ~; s
  
• 
  % S6 s7 n! x, b# @2 |0 I
  say hi 2
  
  ; v' f9 R( K3 @( j
• 
  9 p2 [' |- {1 T, J' w* e
  say hi 1
  7 ?! t, v: T$ w' U, `# C
  
• 
  , O7 \$ b+ K+ M, R' `* t  G
  say hi 0
  . t5 g3 q3 `9 V  x: S! x0 K
  9 _* D, V2 O; B3 V& i
  

3. 进程池Pool类

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

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

进程池中常用的方法：

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

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

• terminate() 立刻关闭进程池

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

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

  
  $ R4 |0 Y# p% P1 e$ {

• 
  
  , ?: T/ Y. B' i$ y0 U) Ifrom multiprocessing import Pool
  
  
• 
  
  6 j. ~8 O, z  Yimport time
  
  
• 
  
  ) o# u6 E) @7 a- ~9 r
  % T4 n$ _! D8 o& M0 o# d6 w. y8 g0 N
  
• 
  
  - h# Q  c& P1 S, S* Y" N( M' @def func(args):
  6 X" U/ E3 {. C6 }. s
  # a* }3 R- w" J# o" i5 |* A# _- X
• 
  
  $ [6 Q. c. n2 X. S% x6 f    time.sleep(1)
  
  7 p7 O! S* g8 M/ V- F& {$ `
• 
  
      print("正在执行进程 ", args)
  
  
• 
  
  , t  c4 U: M% Q  N
  ! P6 R! `" j5 N1 {
  4 K8 A2 P6 g! c, c  e( }3 J6 F
• 
  
  9 F" P4 N, Z- l* N8 xif __name__ == '__main__':
  4 E5 Y! W1 Z) Q% _
  ) R' O, u2 a- Y
• 
  
  5 U& C! ^! K) o. S( e5 e, C" y
  - B( h' J; ], x  [
    `/ {5 a8 H9 k
• 
  " O4 y0 _5 f& |% u& |1 Q6 O' R
  9 B- D* T1 N7 K+ R    p = Pool(5)     # 创建一个包含5个进程的进程池
  
  
• 
  
  2 h5 B% e  x/ p
  2 Z7 q4 _/ Y: o, ?' }+ R  p
  
• 
  
  : `# r+ a+ h; O- T. d0 R    for i in range(30):
  ( H. ?" f  A+ }( G
  
• 
  
          p.apply_async(func=func, args=(i,))
  % K3 p6 q/ u' i* g3 O0 F5 b
  
• 
  
  ( d0 M: [. K# X: e$ n% E8 x
  4 w# N: [  r4 ~0 p5 F3 l" o# Y
  
• 
  , K; p+ Z, `: e/ c2 v/ S5 Y( H
  % c% |- d" D+ z- i0 i, e    p.close()           # 等子进程执行完毕后关闭进程池
  
  4 n: j" p! q1 s* z% i, i. x
• 
  3 |2 i) c- \4 g5 R
      # time.sleep(2)
  
  
• 
  8 N/ [7 U6 ~6 o( S
  ! p: Q$ a: N1 D4 ?7 w- R    # p.terminate()     # 立刻关闭进程池
  
  
• 
  
  ' a8 [  G! }( I  t( _    p.join()
  
  . v- E. o* a) \( Q" R
  1 w* C( r/ A* U" f& C8 o* E

请继续关注我

# ]: r: x  g5 p0 N1 ]# s. Y
- \6 ?4 ?$ _9 n. I, W, U& K