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

杨利霞

爬虫（七十）多进程multiprocess（六十一）
; x- C* y& s$ q5 S

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

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

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

• 
  
  8 u( F8 p: X+ @7 o- {+ I/ aimport os
  
  " P8 J5 k1 |) c/ `- ]0 [3 v5 F
• 
  ) F) M7 Z( M! }$ }% [3 D. W
  import multiprocessing
  
  
• 
  
  - x9 R% W) i; ]- P0 O& V- }# b
  ; s  H$ r0 @7 ?! M9 t% v% f2 o9 n
  + w8 j7 P; I" x- d" H& ~$ O9 _: q
• 
  
  7 X9 }# p1 o$ c+ bdef foo(i):
  
    y* ~+ L) @# U5 o
• 
  & G4 R( T! R9 T3 k% G  e* h
      # 同样的参数传递方法
  + {/ i1 u) I7 l8 F0 [- C
  % q+ `  q7 G; ~4 p0 h) W3 {- D
• 
  
      print("这里是 ", multiprocessing.current_process().name)
  ; r. J0 P6 D" b% B2 E0 }
  / j/ ~' \4 k2 G$ l6 r
• 
  
      print('模块名称:', __name__)
  ! l1 p* ~7 ^" a; y: `
  
• 
  2 M* I1 v0 H7 D  B- z9 w" G
  ! x# H6 x9 R- d9 I4 a" n: s    print('父进程 id:', os.getppid())  # 获取父进程id
  
  
• 
  " S% Y+ N4 S) r- ]+ ?
      print('当前子进程 id:', os.getpid())  # 获取自己的进程id
  
  : F0 c1 A! {3 ~& F  I, `
• 
  $ V. R9 ~6 t) A8 N% O
      print('------------------------')
  8 H4 B* A2 R5 z0 i2 [- i' Z
  . l; k+ O5 x& T
• 
  
  ; J; z* W( a) @
  - Z* M  Q0 N5 z+ y
  % d# @5 A) r% K$ @9 N3 H$ R
• 
  3 H- ^4 ?7 x/ Z: I
  / ]# v- n6 U8 h$ |0 \! @if __name__ == '__main__':
  4 z* C$ h- S1 }6 A) V/ j5 B
  / W7 T& [( M, N% C8 Y2 \' c3 H
• 
  5 v/ D7 {+ X: a3 C; Y6 m
  
  7 [0 a4 I: M9 B% ~& o+ p) D
  
• 
  
      for i in range(5):
  0 l. ^. F) _/ D+ [9 x0 E- F( ?9 w: G
  3 k5 I- g  `, z. P% }& w, _1 @6 o
• 
  
  ; h1 B) A1 g1 C  U9 f$ b$ \* ]        p = multiprocessing.Process(target=foo, args=(i,))
  9 D7 s* y% B$ [$ I' m" I
  " D) U( y0 U# r4 _6 U: E! s/ y7 P3 \
• 
  4 h! g9 k! F/ ]$ W
  ; [. \& K) z- n  a        p.start()
  
  
  - G' n6 a3 b& s

运行结果：

• 
  
  这里是  Process-2
  7 O2 U, y7 C  D4 `9 f
  
• 
  ( `2 [0 s) }. U! f& H
  模块名称: __mp_main__
  
  
• 
  0 {! Z6 `$ q3 I# ~0 a
  父进程 id: 880
  
  3 R  c2 M/ _8 L4 q/ ]" [2 o
• 
  
  当前子进程 id: 5260
  ( d0 ~- w, M% }" A7 m  y8 G
  7 w8 G5 v+ J; O
• 
  
  --------------
  
  
• 
  3 m0 D/ K; R' K# p% f! P
  这里是  Process-3
  ! g% l# S! d2 h
  : y0 S5 n9 W0 ^4 M  B4 ]7 q: H
• 
  " k6 F  M! O7 W# _$ C1 l
  % S9 r0 {. [* k& e2 y! ~* ^' f4 ?模块名称: __mp_main__
  9 V5 t* d* g% W  V+ }
  
• 
  / \3 i& u0 \' v9 K8 n1 Q$ q9 f" E
  父进程 id: 880
  
  
• 
  
  $ U; r7 i$ @$ `  q, q$ P/ Y6 e当前子进程 id: 4912
  
  , o7 U( A9 k2 V4 F6 N
• 
  8 R6 a" A/ F, Y& W7 j
  --------------
  
  - P5 O  s7 `7 M: U/ f: `+ T2 O
• 
  * a: l( D# V7 X
  这里是  Process-4
  
  , A5 Q6 I3 D( G6 ~0 w
• 
  " J' V: P* {& \% ?( w0 r8 L
  4 U0 o/ |2 P1 ]  p& h模块名称: __mp_main__
  
  * \4 Y+ `8 s# U* R, }* W9 k
• 
  " _- Y5 t2 |& u8 K  Q
  / C3 ~# r$ A4 d$ B' A6 P8 ^1 o父进程 id: 880
  % @( m9 x+ b  X: r/ l
  3 K2 t9 ]2 |1 |* }* M: ^# S
• 
  
  当前子进程 id: 5176
  
  ' S6 c: Y7 G2 u- h* U+ _
• 
  & N; W+ p  p. g7 y1 M
  --------------
  
  
• 
  
  这里是  Process-1
  : a5 d# \: O- J& B9 h; g, q
  ' Z6 z1 K8 F2 Z9 N
• 
  5 S4 z5 c- h7 Z5 v1 G! {9 m
  模块名称: __mp_main__
  
  
• 
  
  " G. ?' D+ b/ i: U- f7 j父进程 id: 880
  
  
• 
  4 P1 r/ {" O4 x; I. c4 C; X
  当前子进程 id: 5380
  ( b4 I! Z/ Q4 i- F+ p. X
  
• 
  ) `. c' u- Z0 p; N/ }0 `
  --------------
  * L" l3 z4 u+ C* x
  
• 
  3 r! q$ T; I  l+ ~0 t. i( Z) ?5 D
  # X1 h3 h. y- R1 d这里是  Process-5
  
  / S# M/ B( I  D6 S2 z- y0 E% n  ^2 S
• 
  
  模块名称: __mp_main__
  
  
• 
  2 ^  Y' W4 A0 h$ t
  ) ]& k. f2 F% ]; w父进程 id: 880
  
  
• 
  ( m3 t- A5 G* D) ~+ N3 e
  当前子进程 id: 3520
  
  ) y# A3 L6 l- e! F% V2 M- e1 y
• 
  $ ^2 Z$ }: R; d: f3 [- R
  --------------
  ) n! `9 J# l1 u! R. Z, d3 G
  - M  @) R4 b) S1 D+ |" \6 I
  

1. 进程间的数据共享

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

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

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

• 
  
  from multiprocessing import Process
  . ]+ Z; c9 W$ u: m* G
  ) k5 J( l& G, b* U7 J
• 
  
  $ N" h1 w8 ~  l; a. [# ]! Y
  
  0 v# K% i9 _/ S+ K7 q9 t" Q- a
• 
  
  lis = []
  
  4 R% L! b- m; b5 N. f% a; ]% ~
• 
  , ~& Q1 a4 j5 z7 G: h  Q* ?
  7 J( B% T  q+ x0 N- d
  
  
• 
  5 X/ J% k) J$ a7 J/ m* b  _
  7 G; p$ R' Q+ n; F" k! H8 o0 k2 x% Tdef foo(i):
  
  6 e' _' S8 v$ r, b) S. y
• 
  
      lis.append(i)
  ) P6 `0 \2 R" T, T3 H
  
• 
  ! N1 P7 o, w6 T+ j! g! h2 s& }
      print("This is Process ", i," and lis is ", lis, " and lis.address is  ", id(lis))
  9 L. q' T! k" {; s/ o9 C! A0 F
  3 b7 n; p8 M9 M9 b* e
• 
  
  $ Z. A6 t+ q* G0 w% S- D2 u
  
  
• 
  - n2 x" F6 @4 `, m
  0 l- J. b2 h% n- M) Wif __name__ == '__main__':
  / u- _* M- m. n! ]  Y
  : P* \" y5 ^. Q/ N* z; n: z
• 
  
      for i in range(5):
  
  
• 
  
  2 S. }3 @# @8 P7 p        p = Process(target=foo, args=(i,))
  
  
• 
  ' ]3 `5 t& q1 a' a2 b
  8 ~: w# R" q0 T- `4 V8 l0 _        p.start()
  
  
• 
  
      print("The end of list_1:", lis)
  ' W' v8 n; d; `8 f
  
  

运行结果：

• 
  
  6 K2 L! n. L# v0 L1 X. OThe end of list_1: []
  
  
• 
  
  * l2 q5 E( b( U. M5 gThis is Process  2  and lis is  [2]  and lis.address is   40356744
  
  
• 
  " g% r# Y- C& K6 R1 x! v
  This is Process  1  and lis is  [1]  and lis.address is   40291208
  
  
• 
  1 @7 ^1 f% v' Y
  This is Process  0  and lis is  [0]  and lis.address is   40291208
  
  
• 
  2 Y0 z1 [3 ]: @, y6 v6 x2 W: F, l
  This is Process  3  and lis is  [3]  and lis.address is   40225672
  ) l% ?. h2 m  j
  - O; e/ K4 s8 I1 b% ^7 z
• 
  
  + D6 p  @6 T* T! x- }) PThis is Process  4  and lis is  [4]  and lis.address is   40291208
  
  
  % L* g) a' y5 T9 U

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

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

1.1 使用Array共享数据

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

• 
  1 ?% J* c8 D! P0 @
  7 j& X; h: ]) w2 M1 S# M'c': ctypes.c_char, 'u': ctypes.c_wchar,
  1 U, |0 }4 d6 X! Y: u
  
• 
  
  + ]) F7 R  v. l. g1 U9 {" r'b': ctypes.c_byte, 'B': ctypes.c_ubyte,
  , j, |1 ]5 F, I  W
  % Y; c9 m. p; r8 l
• 
  2 R7 O3 r" P7 [2 g
  'h': ctypes.c_short, 'H': ctypes.c_ushort,
  7 T  Y+ [# Q1 r! _- B
  * e: B3 B' q; R. ~4 S* w) P' y* @
• 
  : d; b6 Y9 e6 H
  'i': ctypes.c_int, 'I': ctypes.c_uint,
  
    w& Y& t3 J% T& z4 ]
• 
  . I$ f1 N  U. S7 ]  F
  'l': ctypes.c_long, 'L': ctypes.c_ulong,
  
  
• 
  
  8 _9 y+ b9 q% b  @$ z'f': ctypes.c_float, 'd': ctypes.c_double
  
  
  

看下面的例子：

• 
  , k$ w+ ?2 _& M, M% k. G; c) w
  from multiprocessing import Process
  " |: s4 P* o  ?0 ]% s' t0 f- |9 y
  ; {' T: S; G1 F1 C: @! q
• 
  
  from multiprocessing import Array
  
  
• 
  1 B' f- ^8 R0 ]  j
  1 u# u+ G$ a* u2 V
  
  
• 
  
    x( f7 j6 H+ W9 ^/ edef func(i,temp):
  1 K, T" N( T. V2 s! D% `; q0 V
  
• 
  4 Y0 t2 g5 a9 f9 D8 A, `8 A
  , u' u: p3 j* G0 `    temp[0] += 100
  9 Y2 t. e6 l3 k% H9 H
  
• 
  6 U# e- ]" Y9 ]
      print("进程%s " % i, ' 修改数组第一个元素后----->', temp[0])
  * W% ~  P. P* U! _
  
• 
  
  ; `# C8 d* X1 o* I" _5 ~% V
  
  
• 
  
  ( Y; _2 y2 ^, P, g7 lif __name__ == '__main__':
  
  
• 
  
      temp = Array('i', [1, 2, 3, 4])
  
  ; Y* {' K; c3 }5 i7 F- y
• 
  
      for i in range(10):
  
  ! t4 M$ Y) |8 \' v4 D
• 
  
  $ _/ _, @( ?- ?& T* ~, {        p = Process(target=func, args=(i, temp))
  
  
• 
  " _9 r9 ]" `3 S/ j( \. b
          p.start()
  : C" e1 ~( y; i3 {, d9 T$ }
  
  : U' v- `' U7 F1 A

运行结果：

• 
  8 a! \7 {  F' m- r( B
  进程2   修改数组第一个元素后-----> 101
  + V$ V7 ~# p: E) v6 J6 j% U
  " E8 x1 }! N' m8 n3 K5 x
• 
  
  " E  v& M. |$ L8 g6 _进程4   修改数组第一个元素后-----> 201
  9 x; Q+ W2 \  T. G6 _$ G
  
• 
  5 ~& U( s1 L7 ^" z" g
  + D6 J. O3 s0 a# ]/ t进程5   修改数组第一个元素后-----> 301
  % F' \8 `" V# K4 o' w
  % ^5 n0 _  G" P4 b" V+ \
• 
  ' d  @. E- }  h+ A5 x$ p
  进程3   修改数组第一个元素后-----> 401
  
  9 P' `0 u+ w! P/ |; T+ x5 a
• 
  2 Y5 ?4 G1 Z2 I
  0 m) k5 W: v, j/ t. r1 o, w进程1   修改数组第一个元素后-----> 501
  
  1 S/ G' v1 U$ h
• 
  8 R. Y, t* K; `  K+ e
  进程6   修改数组第一个元素后-----> 601
  ( |; E) ^4 V! I/ A  D
  
• 
  , @, l) ^0 X0 b5 K! C
  ( [: U+ O0 `9 X进程9   修改数组第一个元素后-----> 701
  
  
• 
  
  进程8   修改数组第一个元素后-----> 801
  - [% Y7 j9 h8 H# i
  
• 
  9 _/ e$ n4 p% h4 A$ d4 W% U
  进程0   修改数组第一个元素后-----> 901
  
  $ r9 U! j! ^+ C- m* Y
• 
  3 l9 W& L4 V3 n! s  @% c2 P' b' Z
  进程7   修改数组第一个元素后-----> 1001
  
  , c- A3 j& ~" x  e: S
  

1.2 使用Manager共享数据

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

• 
  ( V0 {+ f) b6 [; |+ |5 D) k( N
  1 X) y- _, m6 {- u" V. Wfrom multiprocessing import Process
  
  / I+ B3 b( ^! k
• 
  $ _5 N* m& ], o% J5 D, R
  from multiprocessing import Manager
  5 o+ r3 f! W1 P
  
• 
  
  # Y9 h" c% `1 ]
  ; P( o4 `- J7 N) r$ U  W
  3 e7 F, u# y5 P! T
• 
  
  . O, H; L1 E" q' k* \3 [; xdef func(i, dic):
  6 q4 W5 @3 e+ ?/ L9 k( {7 C, X
  
• 
  
  / p4 R0 z' ~2 l# L8 L$ {3 g: i- W    dic["num"] = 100+i
  
  
• 
  + [) E0 D" {6 B
      print(dic.items())
  
  9 W% f+ A7 Y8 [% O& g
• 
  
  " K2 S5 Q* ?; Y+ A( _' c% n- v
  1 |3 c) M  c. X" `
  0 p+ U" r- K/ z, I* V1 \8 B
• 
  , W' ], E) U0 F. \& H! s2 C
  4 o. G, C- t/ ~+ ?6 E  J) O1 S" \  ]% dif __name__ == '__main__':
  
  ) b$ I5 I4 j& Y$ s
• 
  
      dic = Manager().dict()
  7 O% m" F* h, v& V4 Q* i9 @. `. d
  0 ?) ]$ ?% Z+ W) G0 z
• 
  
  ) w. W6 a- S9 Y2 ~& D    for i in range(10):
  
  
• 
  
  " b7 Y3 A# \+ x' x5 F1 w3 C2 u) y+ M        p = Process(target=func, args=(i, dic))
  
  
• 
  
          p.start()
  * y. h3 x. E6 k- g/ V
  , W( \& R9 e& I7 g
• 
  3 f2 f: Q6 w) n* X9 ]6 c7 ?
  7 [. _. h% E# G        p.join()
  
  
  

运行结果：

• 
  6 S$ j0 F' B& W, Z" m
  3 z* Q1 J3 ~. ^& |( s9 P[('num', 100)]
  
  
• 
  
  [('num', 101)]
  8 u3 P/ W* n" ]# G* T
  ' ~7 |- ]9 ]( A& Z
• 
  
  [('num', 102)]
  
  
• 
  
  [('num', 103)]
  ; P# C; W; D+ N
  0 }) a9 c- r2 @) _
• 
  8 K% k* j0 y+ i  p9 e# d. c2 n7 v
  [('num', 104)]
  
  
• 
  % ~+ u  t  W6 C$ Q* h
  1 K% X- X& ]% W+ G: P[('num', 105)]
  3 G5 E3 i: d' D' m' k  Z
  
• 
  4 t. i0 s$ G& J% r. \9 w3 m$ p
  * O& {% m6 v$ \[('num', 106)]
  
  ' h- v! m3 y8 ~- ]4 S7 ~+ L
• 
  & |; b0 c9 q2 L. z( D; L! T
  % d3 @- Y$ q" d[('num', 107)]
  
  
• 
  7 N9 v" U) K  s0 R% H* t! {
  ; r7 H5 Y- B$ d0 G[('num', 108)]
  
  
• 
  6 a/ K6 G* N! m. d& |4 Z: G6 Z
  : ?3 \5 I# ^! P6 q7 Z  c[('num', 109)]
  $ K* ?7 t" ^1 b, r2 B
  5 j4 L9 s5 t- `7 `5 q$ M
  

1.3 使用queues的Queue类共享数据

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

• 
  , L: v$ n; ^2 D4 T
  import multiprocessing
  
  
• 
  
  ' k- l4 ]3 U+ o. Q6 ~" Lfrom multiprocessing import Process
  3 ^9 f* `6 I9 ?1 j# t
  " F0 v/ V/ r4 a& R! v
• 
  " T5 B5 ?: L# _$ S9 f* M
  from multiprocessing import queues
  
  
• 
  9 K! C) ~. c* h1 U' a
  
  
  
• 
  
  + t+ U$ a$ r+ o; @def func(i, q):
  
  
• 
  + r$ y' P$ t1 ]' x( x8 A
      ret = q.get()
  + A" }! m% ~: t1 k
  / ]0 q8 p6 a4 L5 M% W
• 
  - I& \7 ?$ P+ n; k' ~
  - M. z0 u& ~& C4 O. M, k9 \    print("进程%s从队列里获取了一个%s，然后又向队列里放入了一个%s" % (i, ret, i))
  2 {7 S! a" U+ _9 x& i. m7 \) n% _
  
• 
  
      q.put(i)
  7 H4 Q/ i0 x! \; h# A* i. P
  8 S/ K% R6 c' I
• 
  2 l1 K- ~5 a" c/ Q1 p2 k( D, X
  : I$ f$ O0 g: R
  
  + j9 _0 S( f  n1 i1 c" z
• 
  ; u  D9 g$ d; Y. b6 _
  if __name__ == "__main__":
  6 {+ ~" d2 ^2 B& }+ T
  
• 
  
      lis = queues.Queue(20, ctx=multiprocessing)
  
  9 Y/ {" q7 C5 [& L
• 
  
    b3 g$ Y& x9 f) g4 W9 E    lis.put(0)
  $ C3 t; I) `2 \9 p& Z2 x1 s6 @
  
• 
  
  $ T7 C) }8 X$ y  h5 Q0 T    for i in range(10):
  2 z& _$ V  b0 y& z: t# d9 m
  ; V( ]& H" p2 O0 T# C
• 
  8 H8 L$ z5 `% U6 g/ H/ J
          p = Process(target=func, args=(i, lis,))
  
  % S" E& `6 F7 n4 B' T% b$ b
• 
  1 s+ H$ \  {2 t3 k
  ) H, V/ Z; I% M# R( X        p.start()
  
  
  $ B" F8 N' W" z4 |" @

运行结果：

• 
  ! N2 V  u( h& w9 D
  进程1从队列里获取了一个0，然后又向队列里放入了一个1
  2 K" M0 R* q; p) G0 \, k
  ; K3 r. R% r* f1 C3 D, `( u
• 
  
  / @) B; m3 f' G( G" Z进程4从队列里获取了一个1，然后又向队列里放入了一个4
  / B( D. V- l) E- ^1 T& u, P* f, r
  
• 
  
  . Z2 L/ K% T) y1 s2 ?8 \进程2从队列里获取了一个4，然后又向队列里放入了一个2
  1 l9 c' d7 @+ h5 N- j
  
• 
  + P1 g: ^5 I8 P" l
  进程6从队列里获取了一个2，然后又向队列里放入了一个6
  % o5 k" B/ o2 }$ L! ]
  
• 
    J" K8 n9 T2 D! J: z
  / [$ u: J, y0 g1 D8 }. ~* z进程0从队列里获取了一个6，然后又向队列里放入了一个0
  ( L0 u! X. q2 k% H* z$ T
  % c& Y7 K) r" l( q
• 
  " u$ e  c* z3 s: h3 P
  ( I  }! u8 |) S( R进程5从队列里获取了一个0，然后又向队列里放入了一个5
  8 v1 A* v: N. {$ u1 j, E
  % ^: K5 x9 k& ]6 j5 Q
• 
  
  9 f& n( x1 r' l) o* w  g* W& m进程9从队列里获取了一个5，然后又向队列里放入了一个9
  
  9 d  `) f1 H& C: F4 u% M1 O8 S
• 
  # R1 g" ^" p7 c$ x+ U$ p% k  ^
  进程7从队列里获取了一个9，然后又向队列里放入了一个7
  , \7 A% b* O% U& a
  3 l4 L% J7 m9 T/ y0 b6 I% ^
• 
  / r  \* y# `7 [- f$ z& W+ b
  ! E7 e& l# W+ G. A% j进程3从队列里获取了一个7，然后又向队列里放入了一个3
  ) E0 ?4 M9 e% a
  
• 
  
  ( ~( ~1 @6 I) `, C' A3 N& I& _进程8从队列里获取了一个3，然后又向队列里放入了一个8
  
  : k" ^* l* `( W9 C$ b& g- D" ?
  + ]: K- ]6 M: j$ A% a4 s5 g3 y

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

2. 进程锁

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

• 
  
  from multiprocessing import Process
  
  
• 
  
  from multiprocessing import Array
  $ C; Q3 m2 R" @' i5 P
  # }8 E( ~* Z& z2 Y3 _' I% ~/ u
• 
  % u4 L% V9 `2 u
  from multiprocessing import RLock, Lock, Event, Condition, Semaphore
  
  
• 
  
  import time
  
  
• 
  . x% q( c* h( n5 r7 {  I3 }. Q- {
  
  
  
• 
  
  + j) L3 n5 M* w0 d9 sdef func(i,lis,lc):
  
  " z- O. j8 C9 E$ T& M/ X9 B
• 
  4 [# X  O8 y$ U  H
      lc.acquire()
  
  
• 
  7 l5 f& ~$ l8 O: J
  - r5 A! L5 M7 C3 N* l    lis[0] = lis[0] - 1
  
  
• 
  9 d) l$ d* @5 d1 ]% c3 l1 d
      time.sleep(1)
  
  
• 
  + G, V2 V- A. o0 {) _5 F3 B2 q
  4 d% q0 I# a4 H- r6 k  M9 g    print('say hi', lis[0])
  
  
• 
  
  / M8 Z, H, n2 l8 Y* t' J0 D& P    lc.release()
  
  
• 
  ' {5 s9 C+ s' l* i1 S
  $ }/ \9 |& ~; t
  0 h7 X; v# U# c! M: C, ^5 v; n
  # q% F3 v$ G9 d% \
• 
  % `0 o) l" }) N. O( j# U
  if __name__ == "__main__":
  
  - H4 R& K( l" y! Q% z: q
• 
  
  $ j5 M) F9 W5 t! g9 w: n$ E    array = Array('i', 1)
  
  
• 
  
    p" ]) B9 C" ]9 |$ |    array[0] = 10
  : I9 ]1 O$ R: }2 Y4 x
  ; _; V, K% L/ T: v. [
• 
  ! x" L" M. A5 y( _0 b1 ~# M
      lock = RLock()
  
  
• 
  . T. Z5 l" N9 E; V
      for i in range(10):
  & c1 P1 F7 b0 h9 k1 L7 D
  ; i3 S; \& [: T
• 
  
          p = Process(target=func, args=(i, array, lock))
  3 a- G4 O; Q5 c6 S
  
• 
  ( A/ O( q) |6 O9 b
  7 W( Z6 |% H6 O1 [, f- N( Y        p.start()
  ) `2 G' Y  ^9 B) D7 Z: a1 g. w
  
  ) m4 ~+ b1 v4 t- @3 T$ F

运行结果：

• 
  ; X+ q0 ?' w; H& `
  say hi 9
  0 p6 M3 `9 w/ y# Z4 v+ l: }
  
• 
  
  say hi 8
  
  
• 
  
  say hi 7
  + L  r! p# c! o5 D1 Z. o
  
• 
  
  . d( v! x7 ], U6 T* r- J( B+ p3 bsay hi 6
  
  
• 
  7 b. O/ ?% {/ d
  0 O) v& U" w3 H0 h# x  r! U3 t) Lsay hi 5
  4 y; K: b1 ~+ `' Z7 J
  ; L/ W) C. Z/ W2 ]* f
• 
  
  say hi 4
    Y0 J0 {: Y; J/ W2 R; L/ Z
  2 _7 t" {" L/ z0 A1 }
• 
  + L( y/ m& _+ A+ z* n* l  l7 v: F
  . h( F1 a  m$ q+ m1 I: M6 C  }7 F, M# usay hi 3
  
  
• 
  ' T1 c% w$ f( ]* ~( h
  say hi 2
  
  5 L7 N. r0 X' l% Z0 Y# j
• 
  
  7 x& J* t( v( e+ ^( {* K2 {say hi 1
  
  
• 
  ; V/ a4 Q8 @1 H6 b0 g1 {
  say hi 0
  
  
  

3. 进程池Pool类

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

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

进程池中常用的方法：

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

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

• terminate() 立刻关闭进程池

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

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

  
  1 E3 E2 e  R: ?

• 
  
  from multiprocessing import Pool
  
  
• 
  $ A! B" R: T- {: D5 H4 c) k
  import time
  
  
• 
  % S2 H: K+ w; p  h4 V
  
  4 H/ i8 e. [' z* V4 Z& K5 d
  
• 
  
  def func(args):
  
  
• 
  
  $ I" o3 j) P+ X' m( G0 {1 D9 W    time.sleep(1)
  
  4 [& W4 O3 k( q
• 
  7 C9 Q' Q% B3 V& ~8 ]
  " k8 ?# N% Q  @/ J    print("正在执行进程 ", args)
  6 K+ Q$ g, O9 t' I9 K! _
  
• 
  
  1 x; N0 c+ U1 F" Y
  
  
• 
  % J7 C0 c' G/ I! h7 _- K
  ! k" V+ O$ I0 \2 S4 Sif __name__ == '__main__':
  
  / z* I- A8 {- S( [  Z! s; A4 C% y
• 
  . P/ w% u3 m3 n1 ~& f1 ~" B- i
  
  . Z4 Q" S! D; S/ O6 C! p
  
• 
  
  ; f2 S0 O( o% n/ T4 M    p = Pool(5)     # 创建一个包含5个进程的进程池
  
  1 B* z+ Z+ w2 J, K* o
• 
  ; d" E. q% q& p& U
  
  
  " O1 c: ?( K) L, Y$ w
• 
  
      for i in range(30):
  
  
• 
  9 R2 N2 G: [9 R: c; d  K
          p.apply_async(func=func, args=(i,))
  
  4 A, f! U1 ]/ A* ~: H7 X6 P
• 
  7 w% t7 k3 {; [4 n
  
  
  
• 
  - ^# V. v* u8 b; G( c2 E
      p.close()           # 等子进程执行完毕后关闭进程池
  
  
• 
  + [* M2 o* E( h6 H1 ^* u
      # time.sleep(2)
  ' w  S7 T  `0 }
  
• 
  
      # p.terminate()     # 立刻关闭进程池
  0 h* L3 \! C$ B3 Y' w6 o8 G1 j, v' C
  , M# @2 m5 G, t5 {, e- l7 g7 z
• 
  3 P+ \5 `  c% m1 l4 m8 X6 o3 P
  + P& \" @; s, E5 m    p.join()
  
  
  

. v# G3 D) k: [' m! ^9 a* E" z

请继续关注我

0 b* X# t9 y- x1 |2 R7 g