数学建模社区-数学中国

标题: 爬虫（七十）多进程multiprocess（六十一） [打印本页]

---

作者: 杨利霞    时间: 2020-5-31 10:35
标题: 爬虫（七十）多进程multiprocess（六十一）

) y7 L* [. N+ U3 a8 c' Q( R: t爬虫（七十）多进程multiprocess（六十一）

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

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

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

• 
  
  . q2 \( |, T1 u! U. fimport os
  
  : S  A+ y& P0 T
• 
  ( U) l+ Z( @/ u. [
  0 M- P1 o6 r% h; {7 Yimport multiprocessing
  * G6 A( T- `6 k1 C# f
  7 s- _5 M$ }& v1 h" M+ K* f
• 
  
  
  
  5 Q: f: S# B$ ^( A- L3 Z. M# e* ?
• 
  9 Z) U6 O7 i5 P( ]
  $ m6 V, B+ E# f. rdef foo(i):
  
  6 v  ^) B, O' Q* V0 O; u# k
• 
  
      # 同样的参数传递方法
  + D$ b5 H0 ~) g) x
  
• 
  1 D% D. J4 l' b- _: e
      print("这里是 ", multiprocessing.current_process().name)
  3 S1 C8 c  q1 L- O$ J
  4 V  V+ E( U0 O2 V
• 
  
      print('模块名称:', __name__)
  8 E# r  D* D: O
  0 ^+ @( w% |, A: n( O" O2 m1 i
• 
  3 b* {) y* Z8 M* p- v! |
  ; ]6 ~3 e6 {9 j" N; i( a5 \& \+ \    print('父进程 id:', os.getppid())  # 获取父进程id
  
  # X) C0 f- Z% g. h
• 
  9 j. S2 [+ z* o2 |( m& D" W
      print('当前子进程 id:', os.getpid())  # 获取自己的进程id
  
  ! _4 t2 d& j7 p0 R8 c  {8 |
• 
  
  1 W. N. d: b5 w  ^4 R8 m8 D    print('------------------------')
  : Z0 e, g2 f* W& ]% y4 p
  # J- A8 o4 j) }6 h* `
• 
  
  
  
  + L* e- W0 n0 Q0 s! e% H6 Q% G& \
• 
  
  if __name__ == '__main__':
  
  - W$ N+ Y) p  @: d
• 
  
  3 W' j, b2 T/ f& c+ q" i
  # `5 ]. ~0 P" E6 f1 V5 w
  
• 
  
      for i in range(5):
  2 h5 J3 W( ]. I$ _" r
  " v6 {; l6 r2 `9 ]' k( B
• 
  
  3 o9 A9 c$ ]/ H9 ~# n        p = multiprocessing.Process(target=foo, args=(i,))
  ) G# z0 t# I7 Y7 c  ~4 L" R
  $ e& Q9 B6 k5 J' v' q+ x2 J7 o
• 
  
  ( d  d# J- o4 e, g4 A. H" L        p.start()
  $ ], A6 D4 I7 s4 s$ j& v
  
  

运行结果：

• 
  
  ! G6 G/ v+ j. H" V这里是  Process-2
  * m& r- g. N9 d
  % I* G: U2 A" B+ ^0 z' u9 c$ t
• 
  
  1 Y+ H+ s3 u3 b模块名称: __mp_main__
  
  
• 
  $ J  x+ T1 I9 {% W$ h& G: B3 l% o
  父进程 id: 880
  ; _' V" p$ l9 y, A
  
• 
  4 ^) Z8 ~9 ^/ u- O- [8 m$ P
  当前子进程 id: 5260
  
  7 v2 `6 a" |) \
• 
  
  --------------
  
  - y% y! A& y/ ?1 _9 @' l
• 
  - z( L. O+ Z* @
  这里是  Process-3
  
  
• 
  % A) F7 v5 J3 T9 w
  模块名称: __mp_main__
  " o- P# [3 X2 r/ H6 U. n; i/ G
  $ |/ c8 t; J) X
• 
  8 k. k/ S) P+ y2 @" u
  ) s# [  W  v$ |: e& o/ l3 S# a父进程 id: 880
  # R1 _3 M: r0 Z; ^
  9 c1 n8 A; Q# g- g$ J7 o
• 
  
  当前子进程 id: 4912
  
  & M1 }8 w- ?1 V& x8 {  T8 a8 h
• 
  / z+ R6 U$ A( e: G
  --------------
  : r" Q% [- B  B; C# P
  * f" X/ `' u, S# C4 t6 ^
• 
  
  这里是  Process-4
  9 z+ f) _. c+ ~) x  X
  
• 
  
  模块名称: __mp_main__
  
  ' i* t6 Y5 G: L/ T
• 
  5 l. @/ ?3 e. g( Q
  9 i6 R# Z5 r9 F2 Y) h# [父进程 id: 880
  
  4 N$ s& e( ~* o9 H# n! w
• 
  
  : R" M1 I  f+ N当前子进程 id: 5176
  
  
• 
  
  . r1 q* I* Y  g* h. W--------------
  ; i2 h9 X, G8 T$ q2 [9 `
  9 N. V4 a5 n7 u3 V! h; r
• 
  / f) i' D+ E" ^  g
  这里是  Process-1
  
  
• 
  ( B; U# T5 B3 F1 u6 f+ o4 o' U
  模块名称: __mp_main__
  
  
• 
    y3 K' s, z; E7 Q
  父进程 id: 880
  4 P% D& S8 O7 D; k8 d/ }9 @: _+ a
  4 J8 b' r. U! e. ?
• 
  
  当前子进程 id: 5380
  
  1 y3 O5 X& l1 m' @
• 
  - h$ O! s& I2 m1 `5 r
  --------------
  ! \, g9 s. J. I# a4 P, d! ?
  3 Z% z  I" M: O& ?$ ?0 [
• 
  & S, A! {$ G. N5 _4 @1 \; v) ~) T, }
  这里是  Process-5
  % }  L) [! J. e) y" u3 I3 l9 i  r
  
• 
  
  模块名称: __mp_main__
  % {, p1 _$ b. I7 @) W
  9 E, v0 l" X$ z
• 
  6 Q) e7 k2 D6 C1 L: q# ]
  + p( D3 c1 S$ ^% m父进程 id: 880
  ) Y/ W2 P8 b/ p
  
• 
  * D/ G6 N8 k& y# D: M- g
  5 {  ]' F6 e  n# F7 B1 ~当前子进程 id: 3520
  9 |9 }" q+ n6 ^7 d  |
  
• 
  ! n# {" V, Z- d6 N$ [) V8 ?9 ?
  - t3 M( ~9 B$ z( d8 j/ E--------------
  
  
  

1. 进程间的数据共享

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

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

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

• 
  
  , H0 U/ S4 h+ H& Lfrom multiprocessing import Process
  ; z' Q1 l9 |8 @0 f
  
• 
  
  
  2 `( }4 x; [0 i- q
  3 O1 H1 d  ?, J2 ~% s
• 
  
  lis = []
  
  
• 
  : z+ X- A  c8 v, y! K3 v
  
  
  * A3 U  F( E6 m; @1 g
• 
  
  4 |; J! v6 k7 |# f3 I- ^8 qdef foo(i):
  
  1 ?5 N4 T7 [$ U: Y4 ^
• 
  
  . [- U7 U) p3 ~. l  G$ _9 L( [    lis.append(i)
  . I" b- h1 r! G6 j6 U. M
  
• 
  
  / P& q; s. t2 l    print("This is Process ", i," and lis is ", lis, " and lis.address is  ", id(lis))
  9 p$ {4 T% x" F+ y2 M
  
• 
  
  0 k+ s8 U( l0 ?. U2 m
  ( s* {2 m' U1 H8 a
  8 G6 `6 D1 P  }: _- F# W6 N
• 
  
    a% e6 S+ Z' X4 v+ p8 Mif __name__ == '__main__':
  : x0 k4 N# E9 M2 B/ K! P
  - D" e" V0 A$ }1 l, O
• 
    A  j' N0 K  G4 j3 o, H/ x% g2 Y
      for i in range(5):
  . Y" G$ Q! c6 y; {. F4 J
  8 ~! d$ X+ |* [6 a
• 
  
          p = Process(target=foo, args=(i,))
  4 A; ]6 ^, s) X2 j- l9 v
  
• 
  
          p.start()
  
  
• 
  0 v! Q( f- z2 A( A( V9 O2 e
  6 I7 V. ?( w  a% K' K. A  u3 L! l    print("The end of list_1:", lis)
  ! k) }1 @. |1 W' w" Q' e) S$ m
  1 k' n: x4 j$ g2 T, e
  

运行结果：

• 
  8 v+ C3 X% n& E2 v8 t5 E6 b& {
  0 O% W( W# C4 a& K) A7 OThe end of list_1: []
  
  
• 
  
  This is Process  2  and lis is  [2]  and lis.address is   40356744
  6 R2 Q. j* Y: k5 W
  
• 
  
  # g/ _1 ^" `" a" S8 q, eThis is Process  1  and lis is  [1]  and lis.address is   40291208
  & H1 C5 y+ \  m9 j
  3 n: Y. O4 o0 E7 {/ w
• 
  " S9 h1 G! [- |8 d8 w! z7 L2 G
  This is Process  0  and lis is  [0]  and lis.address is   40291208
  
  
• 
  
  This is Process  3  and lis is  [3]  and lis.address is   40225672
  
  ! c6 E* h' @) k: F8 z$ f7 c
• 
  
  This is Process  4  and lis is  [4]  and lis.address is   40291208
  
  
  

可以看到，全局列表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,
  , ?! e$ U/ N) f( p7 g# n0 I2 j
  
• 
  
  'b': ctypes.c_byte, 'B': ctypes.c_ubyte,
  : P$ \: f9 ^# c: H: r
  
• 
  
  ! T8 m, R  L/ v8 h  [. L'h': ctypes.c_short, 'H': ctypes.c_ushort,
  
  
• 
  9 P# o5 W( I. }7 M: P9 C4 F! ?
  'i': ctypes.c_int, 'I': ctypes.c_uint,
  1 T8 Y; e# N2 A6 R
  2 {  ]+ E, ?$ K# u$ b+ r3 V
• 
  
  4 c: V6 D! f) \; K: N'l': ctypes.c_long, 'L': ctypes.c_ulong,
  
  
• 
  2 J. l8 T. |, [
  + y" k3 O, B7 W" A4 p3 Z'f': ctypes.c_float, 'd': ctypes.c_double
  / e1 G5 U% z4 s# z4 l
  
  ; [$ T: p1 H* }" f. Y

看下面的例子：

• 
  
  from multiprocessing import Process
  
  
• 
  
  + ~6 Z" {# K9 C) D' ffrom multiprocessing import Array
  2 L3 L8 H2 A) [% _
  6 y5 w* Z1 d4 a& Q, G
• 
  $ h* Q' O; X# q: }' _) l: I+ |
  
  
  ! Q' v3 q4 J1 ~6 m1 e
• 
  $ V7 m+ t4 K. m3 j/ W
  def func(i,temp):
  1 N7 G( W* U5 ?9 n" w4 M
  2 W- t( S/ y! s0 B" P8 k% x# w
• 
  ' g1 j6 B0 u* t
  5 N5 p7 b# a. G* \$ l) F6 C, ]    temp[0] += 100
  + \$ n3 b: O! J! U, i0 Y* E
  
• 
  
      print("进程%s " % i, ' 修改数组第一个元素后----->', temp[0])
  ! C2 ]7 i5 R4 n
    z* g, s$ H( m/ l
• 
  
  . E4 j: i# @9 m1 I" M) N
  ; H. w- M; v9 h; u
  6 c: ]! S' F* o
• 
  
  if __name__ == '__main__':
  $ O) b: d* K+ y5 Z' W% a
  
• 
  
  5 Y# `3 B+ A9 h% e4 e% f$ ~7 F    temp = Array('i', [1, 2, 3, 4])
  ' o; u2 T1 \9 @
  
• 
  
      for i in range(10):
  / ]+ V/ N# f" Y. s& ^
  " ~" I1 l2 }, z. e  d) e" r
• 
  2 j4 K1 o) \0 u. k
  $ i9 z! h' C) l9 h8 w        p = Process(target=func, args=(i, temp))
  
  
• 
  ) _0 _7 b) `7 _' V. ^. y! g# p
          p.start()
  
  
  3 X9 R( W' h3 \; t# {8 o

运行结果：

• 
  3 x$ ]) ?4 N2 B( D2 j) X
  进程2   修改数组第一个元素后-----> 101
  
  
• 
  ( L8 D2 ?* S. ?2 A! o1 S5 V2 w
  进程4   修改数组第一个元素后-----> 201
  
  
• 
  1 \2 a/ B" ?. B0 M3 g9 P
  进程5   修改数组第一个元素后-----> 301
  7 U2 x$ K7 a" F# q& ~
  
• 
  - r7 Y; M: H; u" v$ e! ?
  进程3   修改数组第一个元素后-----> 401
  
  - L$ @/ o- X# ~6 \
• 
  
  1 Y  k! t/ z0 U* E进程1   修改数组第一个元素后-----> 501
  
  
• 
  
  - J6 @8 G, G) _5 h进程6   修改数组第一个元素后-----> 601
  6 c5 W% N% f8 v
  
• 
  
  $ l5 n' _2 o* R& a+ a" K: [8 H1 s进程9   修改数组第一个元素后-----> 701
  
  
• 
  
  进程8   修改数组第一个元素后-----> 801
  9 z8 H% _8 b" M7 f" I2 {  F8 W
  
• 
  
  进程0   修改数组第一个元素后-----> 901
  
  3 ?, ]  o, ^$ Q7 B; V
• 
  ; B. _! I7 ?# X; g5 c/ C0 S3 I
  进程7   修改数组第一个元素后-----> 1001
  " c. w; M% {; H! e% P: `
  
    c! e4 y  G( v( F! r7 l4 B

1.2 使用Manager共享数据

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

• 
  0 G* b& l0 V6 }  P
  + N3 v+ `& W: afrom multiprocessing import Process
  4 w6 W9 d9 S# J' x/ O" y( C
  
• 
  $ f: B% U) H3 y5 U+ F$ {6 h" V
  from multiprocessing import Manager
  
  
• 
  " R, d3 v5 B0 d! ]
  5 @" @+ _& n8 L% M9 x8 i" A5 D2 @
  
  
• 
  
    ^# y  e" V+ l+ Z3 Rdef func(i, dic):
  
  
• 
  
      dic["num"] = 100+i
  2 R9 [7 k, o9 {6 L
  
• 
  0 O: L& v4 x2 a1 H" g4 ?4 N
      print(dic.items())
  
  $ l+ ?+ K6 V7 X8 M& S( ^
• 
  + a' L9 A/ m+ K8 q$ r
  % s$ P% g; Z6 A. k+ P* J% h1 t
  
  5 ^( T6 R$ l9 X9 H8 l( I
• 
  
  # c" L% P9 o1 x& Kif __name__ == '__main__':
  4 O1 f  B( w  Y4 \+ c0 P1 g
  
• 
  # C+ O0 a( O/ W
  , _* c& H8 b% N! @  D" z    dic = Manager().dict()
  
  
• 
    C+ @# D- H+ R& V8 z! I9 Q, V6 y7 O
      for i in range(10):
  * j2 f8 m; U0 M5 I
  
• 
  
  8 {( x9 S* q* l4 |5 U        p = Process(target=func, args=(i, dic))
  , ~( ?: I5 ^4 e% d' M
  
• 
  , I( n" k+ C9 X& n; ^  ~
  8 @0 D) b9 S; M        p.start()
  
  
• 
  
  8 e. ^# ^1 r+ M6 [        p.join()
  
  " U& x5 V6 P: c0 u' ?& X) l, j
  

运行结果：

• 
  & Z  C2 n+ f; a4 z% S
  & C" l4 ~- h  P$ X) X[('num', 100)]
  7 X7 R/ a% A3 @3 E# X+ [
  & F  p3 b; z1 I, I; f5 \
• 
  
  [('num', 101)]
  
  
• 
  ! J9 b! t; _1 B5 r6 T* i0 w9 n
  ' o1 I0 G, Q* u- v9 U" t[('num', 102)]
  ' T) Y2 \* K* I2 t
  
• 
  ! O8 g+ e3 J* r) t
  6 g- u* R- m- ~/ @7 H, X& n[('num', 103)]
  
  4 ~3 O8 Z( p: E
• 
  
  : G( x0 x0 C' e[('num', 104)]
  % h0 G0 O' O+ a& c
  
• 
  , U4 W# X; M: m' ]6 s
  [('num', 105)]
  
  * _- S; |; L- Y$ r0 u
• 
  ; ?# Y9 a/ \, c9 ]
  ; F  |& u/ M& z+ @# ~[('num', 106)]
  
  
• 
  ; G, O9 ]# v+ j
  + c% p2 @* R6 ^. R* i! ?[('num', 107)]
  2 E1 u* p. [  b- X8 }3 {  I
  
• 
  
  , X  [# z& c2 q& u) W: v. y[('num', 108)]
  
  
• 
  ! j. Q* R% h( n6 x$ C
  - Q2 j' n# I6 v) {. z5 s, p8 a, m[('num', 109)]
  
  0 P4 f) |1 q/ `% a( s
  # J! o8 x1 h& P: {" U

1.3 使用queues的Queue类共享数据

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

• 
  
  import multiprocessing
  ! M! I  j1 o  D$ x
  
• 
  / l! ]* p: `0 t* s& Y" ?
  from multiprocessing import Process
  
  
• 
  
  from multiprocessing import queues
    S/ H+ \+ ^$ L
  
• 
  
  
  
  + U( _& f& X% j1 g- @4 d1 l' \
• 
  
    h( g- T) m1 k- R( g/ Ydef func(i, q):
  
  & W" v% E7 Q2 `, L) C+ d
• 
  
    s: B5 c4 w7 I! E    ret = q.get()
  
  
• 
  3 A/ \& Q. o* @) ]  e, V
  , r5 P. y, U, N, y    print("进程%s从队列里获取了一个%s，然后又向队列里放入了一个%s" % (i, ret, i))
  4 C' T) o% p% W
  
• 
  . q" y  I# X5 c9 e: G. r
      q.put(i)
  / g) y# F8 H" \# {2 w$ Z8 {
  ) I* W; F3 ?. w9 f
• 
  
  
    u" N4 C( K" v5 R; @4 l
  
• 
  ; v$ O8 c- q5 Y/ {: O
  9 w3 p9 N9 x. I* Q2 d0 A8 r0 Fif __name__ == "__main__":
  
  : u8 T" F* c/ y5 G1 R! x* G! e
• 
  ) l% L8 Q1 U5 n. t; o" S
  # d- r# T+ o* V6 q3 Z8 Z4 f1 p. s1 W    lis = queues.Queue(20, ctx=multiprocessing)
  
  5 l  a6 M+ g0 ^
• 
  % E1 a# k7 j* p! M8 Y/ Y! Q* e
  , J2 e8 H* Q: x- _, D: p- k5 o    lis.put(0)
  & a* j( F6 m) s& n* i2 [& Z
  
• 
  
      for i in range(10):
  8 z6 ]: P, e5 K; i7 B8 c* l
  9 K: [9 ?; U. E6 }- ]" `' N3 f+ Z
• 
  5 B# u" V) Q5 w: ^, o  P) a# r2 q
          p = Process(target=func, args=(i, lis,))
    Z2 J4 J0 t. G! V% C3 V' `
  : |: e; {8 j: q8 Q5 J9 R
• 
  
          p.start()
  ! u) \% {& p* _& h
  " l$ Q" [7 q$ D0 n. f& o
  " M6 u1 p9 F: e6 s6 a. L3 d& B

运行结果：

• 
  & O2 V4 I) S& D+ d+ Z& ?/ |: E
  进程1从队列里获取了一个0，然后又向队列里放入了一个1
  
  9 L" ~+ O& _( w" U
• 
  
  进程4从队列里获取了一个1，然后又向队列里放入了一个4
  
  
• 
  
  : v8 [8 o4 B7 p. S/ f进程2从队列里获取了一个4，然后又向队列里放入了一个2
  ( {  J" k" U1 o- U" X
  
• 
  & d9 B6 k/ E; ]# l' c& }" n0 X
  - Z, @6 x& u' Q! ?, L/ m% X进程6从队列里获取了一个2，然后又向队列里放入了一个6
  , B8 k/ a: i% T0 h. p
  
• 
  
  进程0从队列里获取了一个6，然后又向队列里放入了一个0
  
  
• 
  
  1 k8 o& l4 k) z1 m" f进程5从队列里获取了一个0，然后又向队列里放入了一个5
  3 }. F7 P  K, d: u) ]
  
• 
  
  进程9从队列里获取了一个5，然后又向队列里放入了一个9
  * X" L* I1 W& O) E( u
  
• 
  
  进程7从队列里获取了一个9，然后又向队列里放入了一个7
  7 R* Q) ]) V) q
  
• 
  
  ( Y# K3 h  z5 O进程3从队列里获取了一个7，然后又向队列里放入了一个3
  ! ]. `  ]% B2 B+ u2 q
  3 i+ p* j" c% j3 f6 K  B; T1 [
• 
  
  0 `, e/ W$ ?+ u9 Y$ O! d进程8从队列里获取了一个3，然后又向队列里放入了一个8
  % t8 P! c. r0 k4 q
  - I  C! r& ~" @+ i+ D$ `
  5 a* k0 O4 ], x7 k- |4 \6 E

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

2. 进程锁

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

• 
  
  from multiprocessing import Process
  4 C6 g, X% ]/ I5 Z3 q4 s" b0 r
  
• 
  
  from multiprocessing import Array
  9 H# i. o: J  h& ]' e
  
• 
  
  from multiprocessing import RLock, Lock, Event, Condition, Semaphore
  
  . g. D! a4 ^, V% j
• 
  : S3 G& V2 n  ?( `" Z
  import time
  
  ) r1 |" N2 V. v) T; V3 [6 ]
• 
  5 f# s. b1 X5 b; h" m, I
  
  
  
• 
  % S. g, v& V% ]8 k
    U+ m/ i& i+ h  y# Mdef func(i,lis,lc):
  
  ! j$ B; _, }- ?7 N2 [
• 
  / w; Q9 E- r) t& c
      lc.acquire()
  * m6 W: H( }4 z* \7 W
  ; B! d( W! v( q* ^0 e" @
• 
  
  1 ~5 |5 \0 d8 E    lis[0] = lis[0] - 1
  1 E* v2 F# X) _9 P5 s- ^
  , C* b% {3 ~0 c# f
• 
  
  7 K2 O$ n' w7 q; i7 L    time.sleep(1)
  
  
• 
  $ D7 P& i* c: w) t" ]5 A
  : R3 d: M# _: G. X! Y- p    print('say hi', lis[0])
  3 \. q4 p; e3 T: S4 w( C, X
  ' A" V& j; `" H, q! o
• 
  2 u; p$ Y6 a' d
      lc.release()
  
  ' T4 W- @6 x- ]1 g) `
• 
  * @0 i0 ]/ `8 T! p
  " o9 a4 \7 ]& E) |
  1 B& }0 R' a% l8 _- T+ r) R
  
• 
  
  if __name__ == "__main__":
  5 P! X% \2 Y+ P$ @1 d
  6 [, y1 F& k: A& }' y/ a; o. p4 C7 Y0 I
• 
  
      array = Array('i', 1)
  + W: g9 i: l( d5 e  q; }3 M% m# y
  
• 
  7 W0 t- m7 }2 b2 L
  9 X* P% V  d3 e    array[0] = 10
  7 T  U6 P1 i. R& x. u' K
  ) k  Y9 v5 d  c" s! V# n! ~
• 
  
  ; f+ g2 x9 V6 e& [7 Q    lock = RLock()
  
  0 O' P& @- P3 i: I% H& T/ o
• 
  : q* K, i, T6 S+ ~5 r3 w
      for i in range(10):
  - a6 e" _9 p2 |1 H4 ^# F
  $ h6 w# i; X9 y/ N2 ?& D, f$ p
• 
  
  * O! n' T: [; |9 I2 H; v        p = Process(target=func, args=(i, array, lock))
  
  
• 
  
          p.start()
  7 }$ x8 Z5 u1 c+ k: n
  
    B* s7 D1 @& _, Q2 w6 d2 `. B; E

运行结果：

• 
  
  / u# p4 d' j; N% j& usay hi 9
  . B$ P) a+ k9 r
  3 y' i2 r7 W" l. s7 L% D
• 
  " S! }+ S$ O9 T$ R
  ! u2 X  B' @/ p* M" Esay hi 8
  " w( a. R- V; L9 w. W! @
  * \( A  ?: t9 ^7 b5 L
• 
  
  - j! p+ ]2 ]* R. P6 Bsay hi 7
  
  
• 
  + t6 D, z5 |2 H9 L
  say hi 6
    {2 V, H1 K2 d8 b! M
  ) @' Z7 k$ b, a
• 
  8 Z/ z- z' Y4 G
  say hi 5
  
  
• 
  $ v' e1 x( b+ s/ x# y$ L
  - h( }0 o: a/ ]5 F+ Esay hi 4
  
  
• 
  5 W. V/ B. |% `- a. N8 R" o# u
  say hi 3
  
  , \4 E- w- U8 A& v
• 
  5 X$ c7 ?  C! G5 F* f
  say hi 2
  ! _# X! F/ V0 X; b) `9 Z
  
• 
  2 y3 i2 {4 ]# a  u! l; I: r
  say hi 1
  : I! R  E; m7 I4 {; ^1 g& g
  5 T) K; r( m7 B# n) Z# k
• 
  
  1 P4 E. c$ D5 j( r$ `0 ^) Qsay hi 0
  
  & B. A5 G( o* Q4 w1 g' [
  # q# ]* b7 S1 s% |

3. 进程池Pool类

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

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

进程池中常用的方法：

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

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

• terminate() 立刻关闭进程池

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

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

  
  5 P: P& F3 `! l9 t, B

• 
  
  ; y4 E# w/ ^! m1 ^$ Ffrom multiprocessing import Pool
  5 j! b" j' n  ], t
  5 w# M% U2 d. W
• 
  . ^2 h* |8 o! e0 J  @8 z/ j% A6 ~
  3 H' M( _5 o/ q$ Bimport time
  
  
• 
  7 @' v  S, R$ g' b0 A; P7 W# @- ~( n
  
  - r8 `4 I( S5 E+ N: j, X
  
• 
  , L& N$ ?: D- l3 h0 L( }
  def func(args):
  1 }' P" k- E: b5 N, X9 @% n
  # j2 t! \6 x3 r
• 
  
  ! R- k9 _* z4 b) U- a9 c; w    time.sleep(1)
  % q2 _8 F$ J' H4 [; m
  
• 
  ; b/ V6 {) S3 Q3 G
      print("正在执行进程 ", args)
  & T+ {$ q6 D* F/ _0 f% C. w
  
• 
    z% ]! d6 t- h# ?" j3 i8 D
  
  
  
• 
  4 D" a6 m% w+ x/ y7 e
  if __name__ == '__main__':
  
  2 f9 Y, e" O" G0 M: ^* K  F
• 
  
  
  
  - W! `3 X0 D7 D) c' y4 D  w
• 
  
      p = Pool(5)     # 创建一个包含5个进程的进程池
  ; j4 _. S- P7 U, b
  + m) v! [2 ^) L# s% X4 L# N
• 
  
  
  
  
• 
  
      for i in range(30):
  ; W# `% ]* P+ f/ ~# H% d$ z4 A
  , \+ _+ J. c. O
• 
  
  " C( a% h4 ^: H5 U. ]0 h; V& n        p.apply_async(func=func, args=(i,))
  , z# i: v" h* _$ B. o
  5 |9 j4 y3 S& ?& M0 C# U. ]
• 
  
  ; S1 g4 t" Y+ U: ?/ W. h9 f+ T$ `0 w
  
  9 t( ~4 P1 ?' T( k$ U, v
• 
  3 A4 A$ A- u9 i
  4 V7 ^! m% E0 j. t+ z    p.close()           # 等子进程执行完毕后关闭进程池
  " B5 U/ {0 q9 I( C6 e3 S
  
• 
  
  3 V. w. t& ~. E" ^7 {6 W% W    # time.sleep(2)
  / M0 {2 [, h# H! T2 E0 V* \  l4 t
  
• 
  
      # p.terminate()     # 立刻关闭进程池
  + U  P5 f" M, X4 ]9 Z/ Q/ {, C
  
• 
  5 V% X9 r" A. _9 l  P
      p.join()
  
  * H7 Y8 z( E$ ?. d
  

0 G6 L* P/ S; I* ~* D

请继续关注我

. v) B1 p4 T  {2 R+ y; P

---

欢迎光临 数学建模社区-数学中国 (http://www.madio.net/)

Powered by Discuz! X2.5