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