第十五章 Python多进程与多线程

简介:

15.1 multiprocessing

multiprocessing是多进程模块,多进程提供了任务并发性,能充分利用多核处理器。避免了GIL(全局解释锁)对资源的影响。

有以下常用类:

描述

Process(group=None, target=None, name=None, args=(), kwargs={}) 派生一个进程对象,然后调用start()方法启动

Pool(processes=None, initializer=None, initargs=())

返回一个进程池对象,processes进程池进程数量
Pipe(duplex=True) 返回两个连接对象由管道连接
Queue(maxsize=0) 返回队列对象,操作方法跟Queue.Queue一样
multiprocessing.dummy 这个库是用于实现多线程

Process()类有以下些方法:

run()
start() 启动进程对象
join([timeout]) 等待子进程终止,才返回结果。可选超时。
name 进程名字
is_alive() 返回进程是否存活
daemon 进程的守护标记,一个布尔值
pid 返回进程ID
exitcode 子进程退出状态码
terminate() 终止进程。在unix上使用SIGTERM信号,在windows上使用TerminateProcess()。

Pool()类有以下些方法:

apply(func, args=(), kwds={}) 等效内建函数apply()
apply_async(func, args=(), kwds={}, callback=None) 异步,等效内建函数apply()
map(func, iterable, chunksize=None) 等效内建函数map()
map_async(func, iterable, chunksize=None, callback=None) 异步,等效内建函数map()
imap(func, iterable, chunksize=1) 等效内建函数itertools.imap()
imap_unordered(func, iterable, chunksize=1) 像imap()方法,但结果顺序是任意的
close() 关闭进程池
terminate() 终止工作进程,垃圾收集连接池对象
join() 等待工作进程退出。必须先调用close()或terminate()

Pool.apply_async()和Pool.map_aysnc()又提供了以下几个方法:

get([timeout]) 获取结果对象里的结果。如果超时没有,则抛出TimeoutError异常
wait([timeout]) 等待可用的结果或超时
ready() 返回调用是否已经完成
successful()

举例:

1)简单的例子,用子进程处理函数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from  multiprocessing  import  Process
import  os
def  worker(name):
     print  name
     print  'parent process id:' , os.getppid()
     print  'process id:' , os.getpid()
if  __name__  = =  '__main__' :
     =  Process(target = worker, args = ( 'function worker.' ,))
     p.start()
     p.join()
     print  p.name
     
# python test.py
function worker.
parent process  id 9079
process  id 9080
Process - 1

Process实例传入worker函数作为派生进程执行的任务,用start()方法启动这个实例。

2)加以说明join()方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
from  multiprocessing  import  Process
import  os
def  worker(n):
     print  'hello world' , n
if  __name__  = =  '__main__' :
     print  'parent process id:' , os.getppid()
     for  in  range ( 5 ):
         =  Process(target = worker, args = (n,))
         p.start()
         p.join()
         print  'child process id:' , p.pid
         print  'child process name:' , p.name
         
# python test.py
parent process  id 9041
hello world  0
child process  id 9132
child process name: Process - 1
hello world  1
child process  id 9133
child process name: Process - 2
hello world  2
child process  id 9134
child process name: Process - 3
hello world  3
child process  id 9135
child process name: Process - 4
hello world  4
child process  id 9136
child process name: Process - 5
 
# 把p.join()注释掉再执行
# python test.py
parent process  id 9041
child process  id 9125
child process name: Process - 1
child process  id 9126
child process name: Process - 2
child process  id 9127
child process name: Process - 3
child process  id 9128
child process name: Process - 4
hello world  0
hello world  1
hello world  3
hello world  2
child process  id 9129
child process name: Process - 5
hello world  4

可以看出,在使用join()方法时,输出的结果都是顺序排列的。相反是乱序的。因此join()方法是堵塞父进程,要等待当前子进程执行完后才会继续执行下一个子进程。否则会一直生成子进程去执行任务。

在要求输出的情况下使用join()可保证每个结果是完整的。

3)给子进程命名,方便管理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
from  multiprocessing  import  Process
import  os, time
def  worker1(n):
     print  'hello world' , n
def  worker2():
     print  'worker2...'
if  __name__  = =  '__main__' :
     print  'parent process id:' , os.getppid()
     for  in  range ( 3 ):
         p1  =  Process(name = 'worker1' , target = worker1, args = (n,))
         p1.start()
         p1.join()
         print  'child process id:' , p1.pid
         print  'child process name:' , p1.name
     p2  =  Process(name = 'worker2' , target = worker2)
     p2.start()
     p2.join()
     print  'child process id:' , p2.pid
     print  'child process name:' , p2.name
     
# python test.py
parent process  id 9041
hello world  0
child process  id 9248
child process name: worker1
hello world  1
child process  id 9249
child process name: worker1
hello world  2
child process  id 9250
child process name: worker1
worker2...
child process  id 9251
child process name: worker2

4)设置守护进程,父进程退出也不影响子进程运行

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from  multiprocessing  import  Process
def  worker1(n):
     print  'hello world' , n
def  worker2():
     print  'worker2...'
if  __name__  = =  '__main__' :
     for  in  range ( 3 ):
         p1  =  Process(name = 'worker1' , target = worker1, args = (n,))
         p1.daemon  =  True
         p1.start()
         p1.join()
     p2  =  Process(target = worker2)
     p2.daemon  =  False
     p2.start()
     p2.join()

5)使用进程池

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
#!/usr/bin/python
# -*- coding: utf-8 -*-
from  multiprocessing  import  Pool, current_process
import  os, time, sys
def  worker(n):
     print  'hello world' , n
     print  'process name:' , current_process().name   # 获取当前进程名字
     time.sleep( 1 )     # 休眠用于执行时有时间查看当前执行的进程
if  __name__  = =  '__main__' :
     =  Pool(processes = 3 )
     for  in  range ( 8 ):
         =  p.apply_async(worker, args = (i,))
         r.get(timeout = 5 )   # 获取结果中的数据
     p.close()
     
# python test.py
hello world  0
process name: PoolWorker - 1
hello world  1
process name: PoolWorker - 2
hello world  2
process name: PoolWorker - 3
hello world  3
process name: PoolWorker - 1
hello world  4
process name: PoolWorker - 2
hello world  5
process name: PoolWorker - 3
hello world  6
process name: PoolWorker - 1
hello world  7
process name: PoolWorker - 2

进程池生成了3个子进程,通过循环执行8次worker函数,进程池会从子进程1开始去处理任务,当到达最大进程时,会继续从子进程1开始。

在运行此程序同时,再打开一个终端窗口会看到生成的子进程:

1
2
3
4
5
# ps -ef |grep python
root       40244    9041   4  16 : 43  pts / 3    00 : 00 : 00  python test.py
root       40245   40244   0  16 : 43  pts / 3     00 : 00 : 00  python test.py
root       40246   40244   0  16 : 43  pts / 3     00 : 00 : 00  python test.py
root       40247   40244   0  16 : 43  pts / 3     00 : 00 : 00  python test.py

6)进程池map()方法

map()方法是将序列中的元素通过函数处理返回新列表。

1
2
3
4
5
6
7
8
9
10
11
from  multiprocessing  import  Pool
def  worker(url):
     return  'http://%s'  %  url
urls  =  [ 'www.baidu.com' 'www.jd.com' ]
=  Pool(processes = 2 )
=  p. map (worker, urls)
p.close()
print  r
 
# python test.py
[ 'http://www.baidu.com' 'http://www.jd.com' ]

7)Queue进程间通信

multiprocessing支持两种类型进程间通信:Queue和Pipe。

Queue库已经封装到multiprocessing库中,在第十章 Python常用标准库已经讲解到Queue库使用,有需要请查看以前博文。

例如:一个子进程向队列写数据,一个子进程读取队列数据

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
#!/usr/bin/python
# -*- coding: utf-8 -*-
from  multiprocessing  import  Process, Queue
# 写数据到队列
def  write(q):
     for  in  range ( 5 ):
         q.put(n)
         print  'Put %s to queue.'  %  n
# 从队列读数据
def  read(q):
     while  True :
         if  not  q.empty():
             value  =  q.get()
             print  'Get %s from queue.'  %  value
         else :
             break
if  __name__  = =  '__main__' :
     =  Queue()
     pw  =  Process(target = write, args = (q,))
     pr  =  Process(target = read, args = (q,))   
     pw.start()
     pw.join()
     pr.start()
     pr.join()
     
# python test.py
Put  0  to queue.
Put  1  to queue.
Put  2  to queue.
Put  3  to queue.
Put  4  to queue.
Get  0  from  queue.
Get  1  from  queue.
Get  2  from  queue.
Get  3  from  queue.
Get  4  from  queue.

8)Pipe进程间通信

1
2
3
4
5
6
7
8
9
10
11
12
13
from  multiprocessing  import  Process, Pipe
def  f(conn):
     conn.send([ 42 None 'hello' ])
     conn.close()
if  __name__  = =  '__main__' :
     parent_conn, child_conn  =  Pipe()
     =  Process(target = f, args = (child_conn,))
     p.start()
     print  parent_conn.recv() 
     p.join()
     
# python test.py
[ 42 None 'hello' ]

Pipe()创建两个连接对象,每个链接对象都有send()和recv()方法,

9)进程间对象共享

Manager类返回一个管理对象,它控制服务端进程。提供一些共享方式:Value()、Array()、list()、dict()、Event()等

创建Manger对象存放资源,其他进程通过访问Manager获取。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
from  multiprocessing  import  Process, Manager
def  f(v, a, l, d):
     v.value  =  100
     a[ 0 =  123
     l.append( 'Hello' )
     d[ 'a' =  1
mgr  =  Manager()
=  mgr.Value( 'v' 0 )
=  mgr.Array( 'd' range ( 5 ))
=  mgr. list ()
=  mgr. dict ()
=  Process(target = f, args = (v, a, l, d))
p.start()
p.join()
print (v)
print (a)
print (l)
print (d)
 
# python test.py
Value( 'v' 100 )
array( 'd' , [ 123.0 1.0 2.0 3.0 4.0 ])
[ 'Hello' ]
{ 'a' 1 }

10)写一个多进程的例子

比如:多进程监控URL是否正常

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
from  multiprocessing  import  Pool, current_process
import  urllib2
urls  =  [
     'http://www.baidu.com' ,
     'http://www.jd.com' ,
     'http://www.sina.com' ,
     'http://www.163.com' ,
]
def  status_code(url):
     print  'process name:' , current_process().name
     try :
         req  =  urllib2.urlopen(url, timeout = 5 )
         return  req.getcode()
     except  urllib2.URLError:
         return
=  Pool(processes = 4 )
for  url  in  urls:
     =  p.apply_async(status_code, args = (url,))
     if  r.get(timeout = 5 = =  200 :
         print  "%s OK"  % url
     else :
         print  "%s NO"  % url
         
# python test.py
process name: PoolWorker - 1
http: / / www.baidu.com OK
process name: PoolWorker - 2
http: / / www.jd.com OK
process name: PoolWorker - 3
http: / / www.sina.com OK
process name: PoolWorker - 4
http: / / www. 163.com  OK


博客地址:http://lizhenliang.blog.51cto.com

QQ群:323779636(Shell/Python运维开发群


15.2 threading

threading模块类似于multiprocessing多进程模块,使用方法也基本一样。threading库是对thread库进行二次封装,我们主要用到Thread类,用Thread类派生线程对象。

1)使用Thread类实现多线程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
from  threading  import  Thread, current_thread
def  worker(n):
     print  'thread name:' , current_thread().name
     print  'hello world' , n
     
for  in  range ( 5 ):
     =  Thread(target = worker, args = (n, ))
     t.start()
     t.join()   # 等待主进程结束
     
# python test.py
thread name: Thread - 1
hello world  0
thread name: Thread - 2
hello world  1
thread name: Thread - 3
hello world  2
thread name: Thread - 4
hello world  3
thread name: Thread - 5
hello world  4

2)还有一种方式继承Thread类实现多线程,子类可以重写__init__和run()方法实现功能逻辑。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
#!/usr/bin/python
# -*- coding: utf-8 -*-
from  threading  import  Thread, current_thread
class  Test(Thread):
     # 重写父类构造函数,那么父类构造函数将不会执行
     def  __init__( self , n):
         Thread.__init__( self )
         self .n  =  n
     def  run( self ):
         print  'thread name:' , current_thread().name
         print  'hello world' self .n
if  __name__  = =  '__main__' :
     for  in  range ( 5 ):
         =  Test(n)
         t.start()
         t.join()
         
# python test.py
thread name: Thread - 1
hello world  0
thread name: Thread - 2
hello world  1
thread name: Thread - 3
hello world  2
thread name: Thread - 4
hello world  3
thread name: Thread - 5
hello world  4

3)Lock

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from  threading  import  Thread, Lock, current_thread
lock  =  Lock()
class  Test(Thread):
     # 重写父类构造函数,那么父类构造函数将不会执行
     def  __init__( self , n):
         Thread.__init__( self )
         self .n  =  n
     def  run( self ):
         lock.acquire()   # 获取锁
         print  'thread name:' , current_thread().name
         print  'hello world' self .n
         lock.release()   # 释放锁
if  __name__  = =  '__main__' :
     for  in  range ( 5 ):
         =  Test(n)
         t.start()
         t.join()

众所周知,Python多线程有GIL全局锁,意思是把每个线程执行代码时都上了锁,执行完成后会自动释放GIL锁,意味着同一时间只有一个线程在运行代码。由于所有线程共享父进程内存、变量、资源,很容易多个线程对其操作,导致内容混乱。

当你在写多线程程序的时候如果输出结果是混乱的,这时你应该考虑到在不使用锁的情况下,多个线程运行时可能会修改原有的变量,导致输出不一样。

由此看来Python多线程是不能利用多核CPU提高处理性能,但在IO密集情况下,还是能提高一定的并发性能。也不必担心,多核CPU情况可以使用多进程实现多核任务。Python多进程是复制父进程资源,互不影响,有各自独立的GIL锁,保证数据不会混乱。能用多进程就用吧!



本文转自 李振良OK 51CTO博客,原文链接:http://blog.51cto.com/lizhenliang/1875753,如需转载请自行联系原作者
相关文章
|
2天前
|
程序员 Python
Python并发编程之协程与多线程对比分析
本文通过对Python中协程和多线程的特点、优缺点以及适用场景进行深入比较分析,帮助读者更好地理解并发编程中不同技术方案的选择与应用。
|
2天前
|
存储 安全 Python
如何在Python中实现一个单例模式,确保在多线程环境中也是安全的?
【2月更文挑战第5天】【2月更文挑战第11篇】如何在Python中实现一个单例模式,确保在多线程环境中也是安全的?
|
3天前
|
安全 Python
python多进程multiprocessing使用
如果你想在python中使用线程来实现并发以提高效率,大多数情况下你得到的结果是比串行执行的效率还要慢;这主要是python中GIL(全局解释锁)的缘故,通常情况下线程比较适合高IO低CPU的任务,否则创建线程的耗时可能比串行的还要多。GIL是历史问题,和C解释器有关系。 为了解决这个问题,python中提供了多进程的方式来处理需要并发的任务,可以有效的利用多核cpu达到并行的目的。【2月更文挑战第5天】
16 0
|
3天前
|
存储 安全 Python
什么是Python中的线程局部存储(Thread Local Storage)?
【2月更文挑战第3天】【2月更文挑战第6篇】
|
4天前
|
监控 Python Windows
使用python脚本来监控进程
使用python脚本来监控进程
|
5天前
|
安全 开发者 Python
Python中的并发编程:多线程与多进程
在Python中,实现并发操作对于提升程序性能至关重要。本文将介绍Python中的并发编程技术,重点讨论多线程和多进程的应用场景、优缺点以及最佳实践。
|
5天前
|
程序员 调度 云计算
Python并发编程的未来趋势:协程、异步IO与多进程的融合
Python并发编程的未来趋势:协程、异步IO与多进程的融合
|
3月前
|
Java 调度 Python
【Python】进程和线程
【Python】进程和线程
30 0
|
3月前
|
调度 Python
121 python高级 - 进程和线程
121 python高级 - 进程和线程
11 0
|
6月前
|
中间件 调度 C语言
8.0 Python 使用进程与线程
python 进程与线程是并发编程的两种常见方式。进程是操作系统中的一个基本概念,表示程序在操作系统中的一次执行过程,拥有独立的地址空间、资源、优先级等属性。线程是进程中的一条执行路径,可以看做是轻量级的进程,与同一个进程中的其他线程共享相同的地址空间和资源。

相关产品

  • 云迁移中心