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