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