/ G, O* x3 i% d4 v
爬虫（七十）多进程multiprocess（六十一）
' F* O5 u) E* P1 c HPython中的multiprocess提供了Process类，实现进程相关的功能。但是它基于fork机制，因此不被windows平台支持。想要在windows中运行，必须使用if __name__ == '__main__:的方式，显然这只能用于调试和学习，不能用于实际环境。 另外，在multiprocess中你既可以import大写的Process，也可以import小写的process，这两者是完全不同的东西。这种情况在Python中很多，请一定要小心和注意。 下面是一个简单的多进程例子，Process类的用法和Thread类几乎一模一样。
4 B- C# a0 a1 j0 b! ~0 J. T) ?' Y0 W% @! T( S& l& m9 D1 U# Y
import os
- D/ @9 i% r, D2 D: Y
. N. \) I+ a2 @! p" O! [
1 {% A- G1 H3 o5 y0 R3 I
9 ~- F6 w8 f, l* Ximport multiprocessing
" M' ]$ v* u0 E3 J: O5 ?# E0 R3 s, q+ V' E- B
- 3 `- \, _" G8 N: h& n
1 }1 s( G( P% X# ]+ m
# Z; [( P1 w" M; c2 _
% _4 e: u: e6 |& V6 Q/ U - ; G! h( { y- P9 v5 W! F
& g; w3 H8 `$ L" ydef foo(i):
1 B0 v- A& |. Z. `! u, A
9 }$ M2 U4 b, l - 9 k, Q" g1 v! z: f; m
6 J$ W6 Y& Q2 \: Z1 z
# 同样的参数传递方法
+ _; s- l- x# Y% t2 I; e1 |' h5 T1 n O/ A _
- ; h3 J. |: X6 |3 \0 J2 J9 S
8 z( l- }8 Q# k- E; l/ N print("这里是 ", multiprocessing.current_process().name)1 Z2 |$ Z* T' @; x! c$ M
" }& b; z7 J' A3 U7 G: J* S - : Y8 l. Y5 \! d9 F0 ^0 l3 u
$ R) ?6 Y m( I7 R- h2 E
print('模块名称:', __name__)
9 `9 J/ p8 ?2 U0 }% s8 S' y( z7 E0 i O
- " R0 E/ x0 Y0 r9 F
9 V- h Z' K g8 Z K, I5 y/ p
print('父进程 id:', os.getppid()) # 获取父进程id
& X( Z: n8 a* O
2 a. @! z, @0 T+ l" R: d - % s! i6 v4 F: [" _( N6 Q
: o2 j" D& D9 O. o; [ print('当前子进程 id:', os.getpid()) # 获取自己的进程id
9 U- S, \+ S( g3 D3 d. g/ Y- q, w A4 l( p* G# ]3 L N
3 }9 ~- P5 R1 H: T8 t& m6 N
& i6 i6 B( x, m" S6 G4 z9 O print('------------------------')$ n& f) e) v, \6 ~
* x* y! F0 S: L& f. x
5 h5 C; L$ G+ {! w2 L5 p2 a/ ]) G: @6 l" l
" p6 {; N+ A& l2 C+ g; X" V% M) R2 h0 u3 t W- r) T- [
% Q2 y* j; D* { J0 L" \& Q* @$ l% I! f$ k: S7 O
if __name__ == '__main__':
. k4 F0 a- k. |- N
% |1 i$ R8 x/ S: W3 q i- 9 N: c# f2 ^$ ?3 f6 y
3 O& K- M8 Y3 V$ B# n% a ]
& j, |3 S7 q2 `7 J
3 f6 ? X) W- D4 |+ Q - ' u1 W8 l& P: @
. Z; B6 m8 N% O, Q2 O9 N
for i in range(5):6 D# U- \0 I1 U' c' y7 ~) C% r9 O
( c V* r, _+ @1 C% O
* B5 N+ a9 f) W7 c9 L3 \# d, l% n' K6 d( K3 q1 J- E
p = multiprocessing.Process(target=foo, args=(i,))- y, ^% h, n$ w+ F$ p+ c2 ]
4 O* _% l- o0 y; k. |' b( Y+ B- |1 M
- 6 X3 S/ W7 J: }
# s# g2 _6 A( X7 d
p.start()
7 R6 S, C6 f( ^5 w1 x
! z ~- [9 j5 x1 u/ X5 S0 l' A3 q/ H( Q9 n p' q
运行结果：
+ ?* W, @# G! K/ }* p. G6 J0 z
% b; V" }6 j; N+ b这里是 Process-2+ G$ [/ X& j2 u+ U
N% P% ^" ^4 d% S: d7 z6 M
3 Z a9 X3 Q( a4 }: V0 v
. v, e3 Y) k0 M# \模块名称: __mp_main__& q6 {1 w! {4 f3 J; ~; U& E
! P: i$ Q: j( n- ! M0 J3 ?3 Z# F b
H9 ]& A, H7 p3 c1 b+ p父进程 id: 8800 ]! e# {- p5 v# \6 D4 a
) }* n$ F( k& A0 f& H
: ?$ Y% k a9 s0 j" s5 d4 r
- J2 {7 n. A4 a1 J# w2 {6 }! C当前子进程 id: 5260( R8 G \2 r1 H
, f6 _% ?# H% r c3 Q* ~& R y8 P
/ t1 o+ `$ @- V. [: T% m# t: Y! B3 o% S
--------------6 G, z. u" ]8 D: y% G! `$ @
) |8 A4 q% n; u5 b- N* H
! M6 a7 J8 o1 [. w4 U% S. A* m9 x8 C+ \) D8 p+ Q
这里是 Process-37 y" B) J8 m: k; [
4 r! F- F$ ^) z3 {' [4 i* L
+ U% b) l0 V1 D2 n' X: |4 b! [- r4 |" Z
模块名称: __mp_main__$ o! ^# ~7 w6 p, u1 ~
/ F" n' { n9 n* o
1 g$ v+ `& a/ o2 p3 h* Y, o% [, g. X" v
父进程 id: 8803 Y; G6 O& y/ @5 c9 A
# ?0 {: W q: B3 o) h( \' Y- ; `' u4 |8 _, x( X1 X% |* O
# O4 C$ K9 Z! y* v0 S当前子进程 id: 4912
: I4 t4 z& z2 F' ~1 P# x' u
8 @2 j5 F. V% x6 Q, N2 E - 0 c* Y/ Y! f3 f( ]
- k! H& [2 q8 K3 y--------------/ x e) h+ V# C3 L( a" n
+ o8 j4 o/ z4 s8 j! i
# s9 j( Q4 F% e3 ~: {6 u! ^6 p0 X
这里是 Process-4
6 J) M& Z' f( J/ H Y( u$ e" e( m$ k& A9 n M% t9 O
- % Y7 z9 l( E8 L, C) Z& M
( _1 J6 R8 E6 [2 r) x模块名称: __mp_main__
/ f3 P2 H9 c. }8 i) c
2 s' Y. z8 s+ U* @
& x! m! I% ~: Q) }
9 ?4 x9 h) Y$ S* X; ~父进程 id: 880
! G% S3 r. J4 E' L* b. R4 w
. A; f+ M: d1 n- ?! [, u, C# b. f- 6 M! W3 |; s9 ^. T+ }- h0 F( x4 q: L( f
( u# ]8 ]+ l8 A' L5 w3 y! t当前子进程 id: 5176 Q" g0 D+ s* T6 V0 L
0 O$ R+ e. ^' X: ^, k: T
% |3 u9 r P% H/ S2 a! }+ ?/ [# W8 v7 l2 a# J" y: ?/ ]3 I0 }) [
--------------
$ U& O* O" e# E2 J% @; D' N
% M2 `2 G* D0 ]! W+ J
7 m+ q( }6 u9 }/ ?! e; Q5 p* K% U3 F# x/ U4 S
这里是 Process-1
! v6 d8 V: J+ w! k
/ e6 Y& o" p1 Z& i
3 u- }$ w' k0 @4 Y" V. I! k
: _# G0 P' l! k4 B8 e模块名称: __mp_main__
" u' M& ?6 U$ |2 O3 s* O
9 M/ `! l$ Q# l9 H8 U/ b" t
; b! H8 D; n. g2 f: D0 D, v9 x% a+ J, ^' X! {
父进程 id: 880
4 Z& R M+ f' S: R6 ^' P' g0 J* J$ v
; r% O0 k7 A0 T/ _3 R" J* ?+ P- Q/ O A8 f) {4 x) L6 s3 A2 p
当前子进程 id: 5380
$ C' B% `8 `( R n1 K9 C3 k% G& ]2 y' N9 [( p! G
3 v% O4 `1 K& T9 d; }+ S* e" Z8 w1 }, a* j1 K% Y
--------------
( S' Y) H& @) @
0 e" e( D0 i+ w/ P- - g7 h6 l& ~8 D0 r% m2 Z) g: @( L
) T; t/ `- F7 c) S9 D% }这里是 Process-5 f* I6 N9 I( G8 j n* k5 y
. v4 G1 N" }/ k! V) N/ ^2 {
- 3 K8 ^. V; U: l' l, A& }( x$ b: y' D* F
% ~ {- ?) u- _ T- F模块名称: __mp_main__
( y+ j. H. D& @5 m, @: c- ^: F* u; P7 C
- x7 O( L: W1 B3 x8 h O
: Y7 k/ t' X2 B, e" ^" z L5 v
父进程 id: 880
# i8 g' N3 K6 ?8 l, m& k/ q6 u0 ~( S6 `
- : o1 X* c# c! _6 \
4 _" Y, A# A* r/ w2 \4 M9 ?5 v当前子进程 id: 3520* b( b1 {7 u6 D; c
% D* x q V- p' ` - ' ?0 k: P/ ?6 n9 R! v+ B2 _
' [! K$ Q$ T: J5 s+ F3 M
--------------! U3 H1 n: W% Y9 g4 T Z
' o# f( N2 b$ l
" R1 j$ z, n2 E; }7 T& `( J
1. 进程间的数据共享在Linux中，每个子进程的数据都是由父进程提供的，每启动一个子进程就从父进程克隆一份数据。 创建一个进程需要非常大的开销，每个进程都有自己独立的数据空间，不同进程之间通常是不能共享数据的，要想共享数据，一般通过中间件来实现。 下面我们尝试用一个全局列表来实现进程间的数据共享：
8 w7 S4 H) A4 I, y6 W5 `( H' R6 A& r
from multiprocessing import Process" T( _7 N5 z3 a. E) e
8 x% ]2 F0 }) ^0 K/ Q# Z: ~- 8 h9 s( y" g; L1 F0 Z
# r) k, v3 G4 M7 v" q2 H
" [, L2 H: F* R8 l) U8 D( @# h* z" s9 J1 D. Z6 j" G/ p! Z; D
- 8 A- d+ J9 t9 S3 d" ^+ g& ~
' ]# w4 z' Y7 E* ~8 d/ J
lis = []
) C& o$ J- H% w. N$ B0 I# h4 C! A9 q) e+ q$ {$ }
3 V3 P7 f- q! m3 X2 V1 e, ?
# ^+ @* V/ t0 H+ B' T5 {+ C$ G' `
: z8 I7 \( G+ t; a/ E( ~! r) k& i/ F& J8 ?' j) }
- 6 J6 m$ B$ T. G
# G' z( ^% I: ~
def foo(i):
' Z8 N2 z; {" q: T( Y3 [3 V7 }6 ]3 Z# n0 D% x% `( w+ s* H* e7 k
- 4 p/ L) W, ?, @- Z: z6 x* o
. S) {( o) ^" g0 f. m lis.append(i)3 S; @: S! `8 Q
6 T8 x `$ _ I
7 J( s8 v3 K2 P( [! z( p
: L* ^( L2 ^- [* j- Z print("This is Process ", i," and lis is ", lis, " and lis.address is ", id(lis)), g4 E5 P" q( R/ v5 p- A
8 e7 J8 F9 X3 i0 |$ ]3 ^
- Z6 m' f; o& R- D9 X& \7 b$ G. f$ Q9 j) h" L% i7 ~
# y% I. R$ M7 y; M% f) H( n- H3 T. Y9 s6 i
: d+ L" q3 F! T/ R4 ~) J. s: h4 ?0 G# A) k7 T
if __name__ == '__main__':1 ~) O5 J. d, L- R5 \; C
1 Z2 S5 p' U7 s- u8 P+ X- + v. i4 h! n$ E/ g. @2 K B( T. e
5 D5 v& y0 ?& l6 e6 Z2 M. J for i in range(5):! t6 [/ D6 f7 E3 j
$ O- e% H) W) G' C8 A
- P3 P7 t/ d4 L, J2 T/ z4 k
. b& t7 r5 q7 {9 n5 ~ p = Process(target=foo, args=(i,))
# j6 s: \# w+ r: Q
( }5 [2 L i0 w+ D - 4 t* m; r6 d- f! f8 C
/ H3 W6 ^0 v: \6 p p.start()9 ~. i7 s! a$ @
- x$ |' d( d( B% F$ f$ h - ! i; O, M* x0 K1 `" r: s. G
0 k! c9 `$ N3 i; q
print("The end of list_1:", lis)# `0 P" e6 _3 y3 L( {9 {
& i- F7 \1 v6 {% y4 k' p
2 o! R( E# r$ d5 z B# R
运行结果：
8 B: q4 d! v4 {$ Y4 T0 E" N7 Y3 t, |, z% K0 }' |* C n4 W
The end of list_1: []
$ ~1 E8 X7 Q8 @% E3 B2 k+ g. b& F( w* d2 Q% m
- ^6 u3 W( ~0 M, p2 Q: H
" z( m( G0 c& H. R m J! b uThis is Process 2 and lis is [2] and lis.address is 40356744
) s" k, c* z/ B5 v* y$ u) G3 _7 f( Q7 k1 a& E: C. W$ J
- " j% q# G+ O& A! i
2 U: E) a2 E4 R# \1 }! V* FThis is Process 1 and lis is [1] and lis.address is 40291208- A O- u# W& E
+ Q% o! A+ g6 u& X3 T1 D1 L
. q% F0 e! o3 J% C
# N- L5 J% D' X" E4 `This is Process 0 and lis is [0] and lis.address is 40291208# N* i6 y0 c) q$ g. v
) f- @5 b6 | _- ) d8 c$ g' L: Q! b6 g
6 ]$ f, r3 l& Q4 NThis is Process 3 and lis is [3] and lis.address is 40225672
+ z# X/ l2 F5 O* X5 s9 X0 s+ B5 N
- 9 a" \. ]5 m0 B. K+ e
4 Q2 d! I' v8 K( ?( zThis is Process 4 and lis is [4] and lis.address is 402912088 w$ u% d, w( Z3 P8 V7 Z9 G8 X. S+ s/ h
) Q& G8 m/ V$ e1 t5 Y
+ P9 G. F$ w1 }0 d7 Q
可以看到，全局列表lis没有起到任何作用，在主进程和子进程中，lis指向内存中不同的列表。 想要在进程之间进行数据共享可以使用Queues、Array和Manager这三个multiprocess模块提供的类。 1.1 使用Array共享数据对于Array数组类，括号内的“i”表示它内部的元素全部是int类型，而不是指字符“i”，数组内的元素可以预先指定，也可以只指定数组的长度。Array类在实例化的时候必须指定数组的数据类型和数组的大小，类似temp = Array('i', 5)。对于数据类型有下面的对应关系： - # b6 x6 m3 C; x
# Z: M9 K' K) E0 C'c': ctypes.c_char, 'u': ctypes.c_wchar,1 o$ R( n6 J: ^7 P$ }
- @- [7 i1 H% o3 H8 g5 B5 G - . X' c' H, _' x& |" n9 F6 J6 A
& o5 i1 o7 }+ f S r
'b': ctypes.c_byte, 'B': ctypes.c_ubyte,, t' O; M* f( k) p) e1 @
. J; L: V" B7 V) | - 3 G- F+ O# j2 X7 n4 V
' Z6 A; l, V- x'h': ctypes.c_short, 'H': ctypes.c_ushort,
1 i( i: T3 L4 K; T- }. M" c$ T' |* w! Z8 `$ S( J- B
- . r( _( V. E+ W1 _
4 E5 B" c$ J9 J6 y6 ]- Y'i': ctypes.c_int, 'I': ctypes.c_uint,
% E, Y/ _! q+ f+ b0 w, h* c: F% n3 r' t/ E; P6 ~) y
- 1 X/ l2 I6 M; N, l- L: I
, o! ]. o$ ^( _* V Q; ]0 R'l': ctypes.c_long, 'L': ctypes.c_ulong,$ }2 p J. Q2 S5 B2 v, i+ U
( M5 G0 ?3 }0 f
R8 o; S X9 m8 V2 f/ V2 A W0 V' r2 ^4 m2 N& u& K6 ?2 D
'f': ctypes.c_float, 'd': ctypes.c_double& U9 G( h: c- ~- j+ \8 D
, i( E. d2 h8 T0 N4 A: F
/ P. u' |! B3 b' ~1 Z# t
看下面的例子：
7 \* U, E" R# z) y3 L5 z& w C5 k
from multiprocessing import Process
) x3 ]6 y. d2 d9 G6 v4 Q
- n2 T. I. z/ a
3 k3 i' L% w( Z; p* ^, J
( R8 U$ F4 \8 Z5 F& kfrom multiprocessing import Array$ _0 V c9 p2 Q7 d/ k& p6 i
, R8 a4 @/ c2 Y. U! z Q- & K5 s# {0 }' u# O: c% T2 x+ e
5 B! Z& j* @0 q8 W
+ P0 | J: H' ^7 ^3 c
. c. ~0 E" t4 e% V$ Y+ x- h - 1 o: u, ^) r+ y4 q' |" t1 x
6 L1 n! a. M. Z
def func(i,temp):
6 e" v" W7 s7 i* e8 [5 ]
* G0 J0 a& M6 j @
+ Z% ]( r5 a' n' O% \8 ?( E$ {" D& E& `6 x0 l; @
temp[0] += 100
9 g/ H4 F. N: S) z. G; v# M
6 t6 E+ q: D! ]) Q! q
) k: S W6 ], J! \8 M: n
( L% ~' l4 L9 c print("进程%s " % i, ' 修改数组第一个元素后----->', temp[0])
6 u6 B. C" x0 P, A
; V) i C' B8 M. C4 V5 x* b$ G7 k* z
4 t6 t1 l" E: n( H# |! C
# F0 t- j$ K: `0 ]3 b& n' G& x4 ~3 h4 ~* K8 ~
( H0 e3 K/ @/ W0 A" j
2 T3 Y( |9 P4 U4 e; G" ]' D% e
+ m) F4 R! R+ x' }5 {- o% z' }if __name__ == '__main__':
- o* q' L3 s! _8 w( T- U4 N# c- ^3 F8 t! s* @! q6 }
* m7 T. v( k7 e9 z# b' L! B8 k4 ^+ v/ ` [. s) z/ \9 ]6 k9 L
temp = Array('i', [1, 2, 3, 4])
?$ N: A! W3 h
) M* A J9 D L5 S; Y+ J7 T- 7 D' Q3 c8 y2 x3 o6 h' j x2 t
1 O: |: Z: f [ for i in range(10):/ g* s; p4 }* ]& D
! b' h& {% r6 i- _" v* z - & E3 o4 h" C' _5 l$ _. v
8 | n& t1 R) ]' u
p = Process(target=func, args=(i, temp))
; n( C- F& K8 a# [& N1 w0 g/ h1 j' p6 |' u. u
1 i3 h4 j$ ~) y+ @7 E4 O
( [: P9 B+ U5 S# T# I p.start()9 ?: Y% j* t9 S
6 s% I! k$ \+ ?- t
! B7 M, w. R- ~3 @0 X
运行结果：
0 v$ S. d* q, ?: Z5 j
U9 X \+ M( v. S3 Y进程2 修改数组第一个元素后-----> 101
3 g5 \2 A) ]# D' h: j2 Q3 f. O
1 R# |/ S& ?7 P7 g* _2 ^) S- 6 c! C7 V4 h( ~6 h+ y
+ l+ x' f, x$ P2 F. U. S进程4 修改数组第一个元素后-----> 201
$ C- A9 k* V8 y- A0 ^: q# M/ |! `. K& E9 h9 |" u+ N% U# I
- * k2 z9 H+ W1 h1 ]% e* S7 _' G2 s
0 ?5 M; t3 X7 n% W! G8 M
进程5 修改数组第一个元素后-----> 301% V$ Y9 b6 \( y7 t: n
7 s( X% ?. |+ E) M) ?, u7 A& ^
- : h# a& G# A$ G4 s
: b9 R; m" {% I5 G* i8 @$ j6 F" E
进程3 修改数组第一个元素后-----> 401
5 ]- M% c) i+ ?: \+ V
4 M" L( P) G2 E2 t' f - 2 X: |" o6 s5 z2 r
# l: }/ V& E" M% R/ Y
进程1 修改数组第一个元素后-----> 501
5 B- E( l' S# a O0 L8 C, n; l* n% ~3 s" J( S! [3 a7 ^7 N
- . I9 K" Z; f) r( S# N/ i: W: g. I7 a
5 ^ `2 F [+ ^- V" P+ H0 {
进程6 修改数组第一个元素后-----> 601
# x; {0 T# W9 O) A, Q( s/ g
. }' }1 W, F8 z
+ q; [; p' I9 T0 G9 F( {/ R" A( ]5 H& r
进程9 修改数组第一个元素后-----> 701 X2 N* ]1 _: |' @& M) r0 j
' F" C3 O* @9 \, h3 y y
. ?# x2 n# t9 n
8 x6 J/ k! X: [ V7 y0 L进程8 修改数组第一个元素后-----> 801
3 ^0 o }; N7 Q6 d
; J8 X: q3 O" _
8 s, x) s3 e8 V! ?, P$ b6 N- T4 ^2 V, b d$ d! A7 X0 L+ g
进程0 修改数组第一个元素后-----> 901
% t( q7 U$ ]; |+ P K2 w6 S
' {4 ~8 A0 k: h0 Z1 D P/ b! @- ' q* J ?. |9 ?! j ^5 c2 i
; ^1 e# u' @$ b8 X进程7 修改数组第一个元素后-----> 1001: ]# y V1 f' h& x P
& B: [+ A$ r" A, U5 b& {
, l+ x3 W1 e7 t* C
1.2 使用Manager共享数据通过Manager类也可以实现进程间数据的共享。Manager()返回的manager对象提供一个服务进程，使得其他进程可以通过代理的方式操作Python对象。manager对象支持 list, dict, Namespace, Lock, RLock, Semaphore, BoundedSemaphore, Condition, Event, Barrier, Queue, Value ,Array等多种格式。
& e3 u Q$ j4 _
: \# s: Q1 L: S. h! Pfrom multiprocessing import Process6 L4 h% m1 N8 Z# I t. J$ B: D
; s6 I; i8 R& P- : V! B4 b8 p4 R
$ b) S' B$ w. K7 f* u( ^
from multiprocessing import Manager
+ @, u4 {/ g" p% a. P$ _+ E9 ~- s/ o5 U2 P
% A* x; E3 N6 O& k7 U% W1 q; e& q' r
, _6 c+ T( ?; h5 K0 |8 x8 {
& w; j& g3 P- q. e/ q& z
' W# m: C3 A9 y* c* c x& {$ c( S* f6 G
def func(i, dic):$ T. }- j; Q( m" b D$ Z
/ d, @" m L& t- " j: S% Q2 q& A% g8 j
9 R( i+ a! C7 S* o5 j7 C! W dic["num"] = 100+i
, C9 E; L" d3 v- e. l4 o6 T. D: }2 @% @' |) Z6 X; i, I7 ]
0 r$ {4 Y5 ]7 u0 b9 O2 @/ S
/ y: h+ u; e9 O. T* M print(dic.items()) m7 n# D: l3 L% W0 u( M
& s* W" U5 }4 s- d( @
9 T) Q; m, A# a2 `& e# Y5 a2 }3 m6 C
8 Q. @8 g3 E4 p9 \& y6 e4 K6 ?; D( P! i# `9 J
0 U6 a* `$ d, N8 v% @3 j8 g/ l" j
if __name__ == '__main__':
0 Q) F0 X$ Y( [* M, ?: L3 H! k h9 \, G# o
- k) t& X, z- ?6 ]
! ?+ C) v! j9 E; ?5 d3 R( K dic = Manager().dict()4 b5 P$ d0 S& L/ H7 b8 B: R
3 W) N5 o- o' {7 x9 |$ H
- 5 t/ @0 \5 E" d% p
: w# r6 T! h5 l g
for i in range(10):
2 |# L! d; K7 Y0 N! Z& f' B) m- I! }' s/ }' V+ U
; y, B7 T. u W
# o/ n, b) N( y2 v) F& t p = Process(target=func, args=(i, dic))- G+ w8 w: d3 e5 {# N
" j( k$ c! Q9 ~1 ^
- {/ D: Y) A# w6 K9 l$ I4 p! I
. U) }0 j: v- V! [ p.start(): i0 x4 v [# \7 K; o
" r! S- {, u: j
( Y, P: f! ]9 ~ n" P6 W2 i8 W4 M6 h3 z! k7 m# C5 H- t
p.join()) n, N' R3 Z4 {
+ j8 X, e+ `6 O' Y) P
. H% F2 Y6 Z9 t
运行结果： - $ F/ [( w# X- h5 X5 I
7 r- ~2 y2 c' J1 w" w0 I[('num', 100)]
: U: U6 h' a( ^" Y1 O* B, y8 x1 \+ H h0 _; @- |
- 6 X& F: P8 G0 n* @
+ \* j. N+ m4 o3 f0 U: u9 b[('num', 101)]
0 y. f& v1 @9 B$ I; w6 ~/ k7 Y b8 T( C% l( ?
- ) f* D0 H( H1 o
8 ~" d* a2 ]/ J$ p% Y6 f9 n[('num', 102)], l0 [$ V" v' l
" W- ~ C( t0 z+ R6 { - ; @9 p/ i; q6 S
4 t) q0 i# y8 c' L1 @ t
[('num', 103)]
& W' h' G8 I" I( z9 T' |4 q9 w
# \ X- m! N' R$ h - " ]1 Z4 o, Q. O( k) s, z- g
, e5 N7 W" W3 E. q& X& ^[('num', 104)]
0 ~0 j' @# C0 `# h( { N5 R
* O+ u6 s$ w! X+ y
5 b" v( R7 ~( [4 N; d2 S9 Y
" S- w; l8 w) g# w% E* W[('num', 105)]. u0 C- d# H% s5 V, e5 E
4 v. R' q+ u; p& N
- 2 K k5 S$ Q+ r A% P' Q" j+ J
9 |" K \9 i8 y
[('num', 106)]
7 A f* ^2 T* O% P0 N; ~1 D. E2 x5 s5 E% O2 Q6 r6 O: ]
- 7 ^! r$ K1 U g% Z
( I1 H. X ]" q# O- N9 l, H2 R
[('num', 107)]
8 L3 z3 d" i, F1 B2 }, x
- k3 `0 ^/ f1 r7 ]+ l - * i9 z$ X1 ~4 F
2 o7 W7 i/ U6 U2 Z4 }
[('num', 108)]
( \/ q3 p; K `1 e2 M! q! M; d$ N) O3 e6 p, |5 z5 |. w
- 6 v/ O h, c, k$ o) ?
# }- V& L1 K4 g' u
[('num', 109)], x- X8 d6 ]" v J3 \6 _
2 g i8 Y' }8 X$ o6 a
" V) o2 Q# j0 ]0 ?) \' A, @
1.3 使用queues的Queue类共享数据multiprocessing是一个包，它内部又一个queues模块，提供了一个Queue队列类，可以实现进程间的数据共享，如下例所示：
! P$ z4 X6 X" y6 o: g: d2 x, _$ k
6 z( l( l' o) @. a2 e3 i. [import multiprocessing, {: t1 ?8 E1 S* G! J
, J9 Y/ n6 @0 Y( J, n- 7 P9 r2 S6 ]3 i: j0 j4 f: v
5 i0 ~" q$ X& h' I
from multiprocessing import Process
. e. n6 g% B. K+ h8 C w) N" r1 g8 v8 v: f% v0 F+ m4 Q
- 7 b. j3 `: q. _+ A3 g4 t
0 l6 L6 F# p/ ~: F( j" P" r. p/ q
from multiprocessing import queues
7 a2 ?) j) p7 H5 }. g8 ]$ t) C( ?# F( j0 ~/ a1 I% g3 m: V6 N V0 K
- $ ~/ N6 Z1 s' k2 S; I
+ x7 G, ^/ p! l+ r6 C# Z8 B8 G' w6 @
+ ~) e8 p; p7 ^+ {; H
# b4 X' [' _/ j( O# X; [ e4 Y! U0 c2 R5 C& |, x% [. a# E, B" j
def func(i, q):
$ A& _9 @) E* Q
2 T4 z$ v. C5 t0 F- # A* t4 V* y8 m7 b7 ?! n
. H8 A+ ~8 k, \
ret = q.get()
* y( G) D0 p2 @3 a4 P( e: y
, f0 _( `& ~, Y& R9 i' l - - {! Z, C, R4 g/ A$ P( P
7 k; f0 \* D0 v! M ~" x3 y print("进程%s从队列里获取了一个%s，然后又向队列里放入了一个%s" % (i, ret, i))
J# s& Z$ L1 ^( r* O% s. E! `
- {9 Q) w! W/ U. C+ @ - ! }: P3 [( }0 r C5 c, N0 p: A6 z* w9 [
: `/ I1 A* ~6 O: `# t
q.put(i)
0 r: w' p2 D4 N* B; a" `: Y, v2 ?* ^( v6 |7 u+ h
- 9 K5 ~& T- ]7 b
7 X7 i% n1 V2 B# r$ h, z; ]2 D) I8 l& Y% w B" M8 }
/ T6 }5 m) n' z8 @7 d
3 P; R d; S+ u/ i: E: U% n# y
2 N9 |, P. w9 c# C6 rif __name__ == "__main__":
9 j: ]' f" H1 C, o! }) J4 h
" C( |1 d" q' i, b8 N
. L% E& {6 @; N; v5 h$ _8 Y
0 i5 D9 |5 A3 k# B4 m# `$ x; v lis = queues.Queue(20, ctx=multiprocessing)
8 x- U" J% U/ K0 `' N- q2 @- C. T/ ~' n) c8 s
- & c5 V* `& Z1 K% p
1 Q- T5 B6 H8 q# k* v! }
lis.put(0)
: @) T/ c: E/ Z- n& b; c( B: ~. ], Y
- / |3 `9 ^( X: E) b5 }% f
& O5 _" d( B0 P; c for i in range(10):+ c# p3 x4 i- M0 _: `$ Z
1 Q" @. {# {: p" I5 E
`1 x S7 z' ]: D* ~8 R* A; X% m( c6 { \( t4 U4 d
p = Process(target=func, args=(i, lis,)): p/ K, F- \4 [& G
* V: U) S/ M& w* E f" u
' q, j# p4 {4 g) N+ x
8 E# q8 W: A, S# s- d p.start()
+ S! y& v d4 k8 y
9 c- V3 k& @2 ]* l* d" o9 ~
5 ~7 [" Q- K) t& y
运行结果： - 6 E2 G' R+ p+ S
/ f0 N; x" L p1 z, F8 C, ]' k0 T6 q进程1从队列里获取了一个0，然后又向队列里放入了一个1' K0 _; e7 `+ I4 A" }
& ^/ y9 g' E) L& k3 t" q
0 Q, m6 C, g& |8 f1 N# i ~3 P! s/ ~2 v/ e5 A0 o0 N
进程4从队列里获取了一个1，然后又向队列里放入了一个4" Y8 {3 N, c5 Q6 k. j% T5 V
" h- Z8 T. S' \$ A
) y4 R% ?0 X, u" d* M. p9 m0 T7 l, F" ^8 ^
进程2从队列里获取了一个4，然后又向队列里放入了一个2) B/ t/ j8 y) D" w' p' I# _
5 L+ W. ^- B# w4 O" b
# f9 s% V" B4 }# J& n# i& D- z5 x
进程6从队列里获取了一个2，然后又向队列里放入了一个6% u( B7 W0 v; @ _
9 m! G; Z) s' X0 M
- 8 ^& |3 r- z. _8 i$ e
9 J1 D4 ^. Q, X1 @; j进程0从队列里获取了一个6，然后又向队列里放入了一个03 b' K, o( @ g" w, I" @
1 T( [, m; n l4 |; d' A: e
/ N, v+ E& J" I+ `& T# f) B* q4 a+ f, Y) p6 O }: _* S6 Q
进程5从队列里获取了一个0，然后又向队列里放入了一个5
d0 z2 Q3 Z: V- ^; g1 v- ~5 }' k$ }" F }. B. r. M, n
3 [ ^3 e R- j2 `5 t
f t7 x3 W6 |. x- l4 Y进程9从队列里获取了一个5，然后又向队列里放入了一个9 C( j1 L7 C1 F3 y
* l" v; M$ a( `# }, p
9 C) B9 V- p# e; n$ h0 O. }
9 @. y i1 Q1 t/ [8 D进程7从队列里获取了一个9，然后又向队列里放入了一个7
0 k% b" C) h7 R; G8 `% _4 U8 [
% f, L+ e. E! N. c
) |. L% |4 j4 q& H6 c" u0 K2 T |) O& L7 b+ p
进程3从队列里获取了一个7，然后又向队列里放入了一个3
( \6 X' [& U% |6 k1 T; X' `4 b* r8 z) a. a( ~
* H, A0 r0 |1 n6 }8 ]5 c6 d& D& f) B6 K4 w7 r
进程8从队列里获取了一个3，然后又向队列里放入了一个84 r+ N4 D3 z# G8 T
7 {5 P# L. {6 R( O) }3 |- N; `
2 \8 F) A( |% K7 _$ M
关于queue和Queue，在Python库中非常频繁的出现，很容易就搞混淆了。甚至是multiprocessing自己还有一个Queue类(大写的Q)，一样能实现queues.Queue的功能，导入方式是from multiprocessing import Queue。 2. 进程锁为了防止和多线程一样的出现数据抢夺和脏数据的问题，同样需要设置进程锁。与threading类似，在multiprocessing里也有同名的锁类RLock，Lock，Event，Condition和 Semaphore，连用法都是一样样的，这一点非常友好！
8 S- u2 x8 Q, y% ~' B1 B
8 d) A. t9 a6 D6 G2 ofrom multiprocessing import Process5 _3 q, v: `; J( P* x( {( Z7 D" f9 H
: E( d( i( P+ C: t
1 v8 t6 |; w ?% g+ }$ c" x. _2 B7 G
from multiprocessing import Array& B9 Z: _# y: }9 K7 W
/ j) U: N0 _ `+ \8 r: w2 D
0 z8 a* n& Q. A1 e- t. Z3 M, T) ^4 |# K7 [' J6 o7 y" R
from multiprocessing import RLock, Lock, Event, Condition, Semaphore
{: B6 @! K$ x! e2 x
% z) Z- W/ Z7 T( t6 Q+ U- / I7 ^. y C1 e h9 H% @3 L
0 `8 H5 j4 x) Q6 p" m' s8 ^
import time
7 L6 k2 P! |' \: o7 d7 o Y& }+ i }; D" T' h1 W) h
# [/ C, P5 d7 d/ F: z$ K( {, n% ~' j: u( E+ i
+ o* J y) `: w+ ?9 e! X& @
: C; R4 j+ n7 ]5 J# t5 J
6 ~6 X$ A) U# }& w& m) z( l" x7 j2 G/ F. Z
def func(i,lis,lc):
* N9 T6 P+ w- t. V; C, ^( u4 U \5 B' }/ F v% P4 L* r
- ; O6 ~. V" C" N% P( R. }# r
# H! O1 f* v; a- z( w' S8 R, x
lc.acquire()
) O8 O$ s8 _- D- G
0 j2 a& x1 \/ d5 U
/ e- h9 q0 }; C1 |6 F% w& g4 p- r4 Z- V
lis[0] = lis[0] - 1: @4 E+ ] r' A5 h0 G8 d) \
% D! v! n; b8 O& T- ( G2 l2 [; m8 o- }6 D
8 ^$ s8 G7 {% y) e; c& v4 L
time.sleep(1)% t7 Z% g" V _( K' n% j, q3 F
% ~0 b! H. x" F$ v7 X8 u* L - , |; J3 D! L0 W( w0 y
! l% M$ f$ R$ Y7 Q# \# {+ Y print('say hi', lis[0])
6 F4 E) S8 K: U" x3 V Y' L. w% m% v7 `1 b
$ G1 _, `2 H% R3 [. f1 w. j/ d: h) G* G) s9 l7 \7 a
lc.release()) D4 V4 O- s* f4 ]5 E; {. t
. U7 B. P5 c9 n( E7 h
- % l6 [# z' X4 w# O
, ~' t: q8 S) i7 ^3 G
5 Y1 ^6 ?; @ \3 {% ]
# d$ @1 H1 z% G9 z4 b) m; u% e
- 7 V6 @5 L8 {. o4 _
- S0 H3 {( a* b# ?9 U3 H
if __name__ == "__main__":( C" e! y' D8 |: f6 m
! T2 Z" N. w1 W- D9 ]9 A - $ a2 H2 i; x7 _3 {! _9 m& \0 M
8 X) f, }; r! ~! r' I' @* u array = Array('i', 1)' W. j1 g) z4 u! q
+ |4 S' N3 I* @" I3 E7 U
/ I3 X. e& C; V9 k
, i1 X" U2 E! h1 \5 |; E- e) j array[0] = 10
# v" i+ B0 T! w$ a9 E1 _* W" U) o2 y8 Y6 e* p
- ( z; n' f& E" Z' M
6 t) U: Y5 Q) m8 C
lock = RLock()
& k/ F# t P, w N3 }/ l
" @( |! N6 Y% f6 [" {" v- r- U
( q" b e% \& W, D5 {5 ?( K! L n6 o* n" F* K& U/ F
for i in range(10):
! I( m' p/ X4 ^8 z( D4 ]3 D- [7 y% E w
- q0 P* K1 f% y/ P1 d; m7 Y- t
; J/ Q1 U# [! T1 k* { p = Process(target=func, args=(i, array, lock))# ?6 a# C# ^' T1 g: Z4 m
+ M6 q* P: m, [0 x/ D1 J
- ' |3 ^5 \7 f7 m6 {( {: Z
3 Q5 W& x) T- D6 H+ M p.start()
7 v9 k# ^$ X' Y- |/ Y9 f( N; G6 F
4 |# m1 K) {' }# z, K5 ?" z
+ h! P# U5 ~. f; T+ S5 i# f" n
运行结果： - ; W8 Z) T7 x) {% y
+ I* v9 Q# O* S4 o8 i7 Psay hi 9
8 k3 S R" K! e+ \ ? h4 I) \6 W7 `3 g7 q, b# M' i
0 U5 A, X1 ~5 c1 g! B. t# F/ @; y/ N6 W+ s' v+ B7 g6 T
say hi 8
1 s8 Q" ~5 E4 N6 U( p; L/ D, L4 H( |4 u9 Q3 {/ G
- * v' Z7 M. _5 d' k2 P# U
9 y1 @3 V3 v; u- b
say hi 74 ?, T" O% W+ ~4 e I# D
$ W! \; [5 m& `# A; s& Z
& U6 ~; v2 W( g+ i7 Q5 A; _* }# `2 v0 {4 I4 z$ Y
say hi 6
5 b3 ?7 v7 S% e( S
/ A/ p6 A# }8 J7 B" _% l0 D; s( @- / _; t( U3 J( Q* p8 }7 A1 t" x
e3 z+ e4 w4 X5 O) S! I
say hi 5: z% B D5 b4 i1 K
& V: z a1 v+ C1 Y, {2 H% q
# w3 _& E3 b' W3 e0 i$ K! n4 E/ y5 w6 B( [/ k' K
say hi 44 z! v8 R8 o! Y- m% F2 Z
' k+ S. ~* }; J9 C1 C. Y- 9 s' y- X- x. e2 F: F P$ \4 f& G
' x2 F7 e$ a3 @6 P t
say hi 3' `% B( F( V, y+ K( K) c, o, O3 R* Q
4 k* M/ b( y) |; @2 c0 d& ]2 Z - ( D. v) A+ k% d% c- P: C4 ^
0 X( V$ ^; O% R Zsay hi 2+ I- w* [8 e3 {
5 R5 P1 ^; j5 X$ r/ G9 b. x - ( W- I( p- i5 U
7 C8 n( v0 B6 k5 Q- \7 ~, e
say hi 1
/ o' o) }* O+ j0 V
2 J* c& w# v; J" ^% m - $ i0 @) O7 B! F8 {
/ W6 R! j% x" y; E [7 z
say hi 0
0 h( w2 D; c3 _# f$ |( l l. l8 t6 ~; Q5 k# n9 w- L S5 Q9 w
8 ~3 ^" t" B w
3. 进程池Pool类进程启动的开销比较大，过多的创建新进程会消耗大量的内存空间。仿照线程池的做法，我们可以使用进程池控制内存开销。 比较幸运的是，Python给我们内置了一个进程池，不需要像线程池那样要自己写，你只需要简单的from multiprocessing import Pool导入就行。进程池内部维护了一个进程序列，需要时就去进程池中拿取一个进程，如果进程池序列中没有可供使用的进程，那么程序就会等待，直到进程池中有可用进程为止。 进程池中常用的方法： - & b3 S& ]; w; J. L ?6 p1 @- O4 D
) B8 _8 t( l1 s; N0 f% P7 |4 a) ifrom multiprocessing import Pool
& I6 `+ J$ n m1 u
5 P* d p5 k: y# ~( u$ X
0 J) c" b r' M1 |- B* D6 d( [; \
1 B; @% R, K8 D6 Pimport time$ V4 |1 D5 E* ? X8 V
& ?. @$ J4 u2 U. {, z- p! u) z- 8 B6 E3 d) M. k, g7 G: _3 p
# D; R- D3 a+ L7 `
- U% b- H* R& W0 r. Q8 I; o+ N6 \1 N7 n) Z
- I% m; j1 \) a+ q1 k: H- [, n: W% u z0 O7 t+ I$ y
def func(args):" u3 r# {$ b* ~3 h
9 a+ R' @ f- Y4 z; p, R% ~- ]
- # B5 D3 h9 m, [4 I
9 _6 Y/ N& z: C( i4 j$ e# U4 `
time.sleep(1)
c7 O: e; u" u- c; ~& C
" B H1 p8 B: K$ M! q5 b/ R
# z! B& q2 J5 |! Y2 h" _! w2 E9 f P' Z/ K2 }
print("正在执行进程 ", args)9 \8 z* b+ ?/ k: H, c; e
7 U* \) k( d" M- 1 m) \" V6 L. X2 S% `
5 \3 N J1 v. B* X1 [# T3 i4 x$ S2 f- N J
+ H: \7 G6 U! a" [9 o - 2 q1 u7 s; w4 `. }: q2 c# C% ?8 d
4 O5 ?' Y9 n" H+ X# q
if __name__ == '__main__':
" `# q1 L* F3 a8 c6 c/ O3 c0 E# G6 ~8 \1 t: J% Y
# n7 t! J' D4 x/ J5 Z2 L' L6 B- R' l# G M
3 M. J* Q3 j' H7 `+ B7 H% T$ T8 J+ I
( s# O- o- ~9 t" o) I$ ~% j- + W) \6 s9 Q: J* ]
3 X8 t9 h$ }2 t7 M' I
p = Pool(5) # 创建一个包含5个进程的进程池
: v+ b1 d& ~" @6 A
! g4 T, ~, z" |$ o$ S8 m" y
2 Y$ y: _% |1 r2 G. r7 h$ P; Y( g$ P' j8 T2 Q; B5 `1 @8 `- i
8 d: s2 {" K- d% Z* o/ |
$ A; |4 j" E9 N4 |" U r- 0 r# a6 \( [! L
* Y& s2 O- G! s' L
for i in range(30):' r7 C2 n# z; `* F. ], n% M
- x1 s1 _: V! O, i* B% o
- @2 R# Q9 @' x6 e: |% X! |- A
* X; Y6 X7 M. F5 k2 M p.apply_async(func=func, args=(i,))
: ~0 r8 y/ ]3 g- n5 v0 J/ U
) |, p* R# p$ l& p
1 Q# R7 j/ ]7 m9 f2 `
; f& o# U1 `: P$ T; l$ t+ K1 C+ y
/ u' j" k( N( c2 i3 x1 m/ L- t
5 w3 \! e" @/ V( @
/ g* u% L. ^& p5 \) K# u
1 r4 i; i( X$ r/ p% ` p.close() # 等子进程执行完毕后关闭进程池
/ E+ E V/ W- @# Y. E
3 T" ?# w8 n- |. E- + T5 u1 t# k# `) h* F' a+ I/ O
- t, S7 T/ A1 p5 j
# time.sleep(2)
$ `+ b& p5 f& e) p$ E
+ u# w5 S/ X- m" K- n5 u4 O6 K
8 V7 R% j! O0 H
( g; G: ?2 z0 p+ u0 n # p.terminate() # 立刻关闭进程池 E. R" {- \* d" T. z
2 R- [3 ?* b5 ?+ B7 E# ?# y& f: @
4 }0 `0 K( Y2 l( L/ C J( K l) O, x
3 V7 Y9 z3 K; i6 M _& W; o O9 N p.join()
* B0 Y$ g0 A- y- \3 B$ b
; f* |2 }; o, y6 p1 b0 }- L, {& x3 T0 t# X- U( T+ H. @; `& H
- {5 p A E5 ?' ~7 i+ i B
请继续关注我 8 e }1 m4 A! Z
" X3 _9 d; H0 Y8 x: U |
IP卡
狗仔卡
发表于 2020-5-31 10:35
转播
淘帖
分享
收藏
支持
反对
微信
提升卡
置顶卡
沉默卡
喧嚣卡
变色卡
抢沙发
显身卡
