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