第十五章 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 id9079
process id9080
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 id9041
hello world 0
child process id9132
child process name: Process-1
hello world 1
child process id9133
child process name: Process-2
hello world 2
child process id9134
child process name: Process-3
hello world 3
child process id9135
child process name: Process-4
hello world 4
child process id9136
child process name: Process-5
 
# 把p.join()注释掉再执行
# python test.py
parent process id9041
child process id9125
child process name: Process-1
child process id9126
child process name: Process-2
child process id9127
child process name: Process-3
child process id9128
child process name: Process-4
hello world 0
hello world 1
hello world 3
hello world 2
child process id9129
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 id9041
hello world 0
child process id9248
child process name: worker1
hello world 1
child process id9249
child process name: worker1
hello world 2
child process id9250
child process name: worker1
worker2...
child process id9251
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([42None'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
[42None'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.01.02.03.04.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,如需转载请自行联系原作者
上一篇:linux内核升级(ubuntu12.04从3.13.0升级到3.4.0 )


下一篇:SQL快速入门 ( MySQL快速入门, MySQL参考, MySQL快速回顾 )