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