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