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