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