多进程
Linux 系统
linux系统可通过os.fork()复制当前进程状态作为子进程。复制时子进程返回0,父进程返回子进程的pid. 子进程可通过os.getppid()获取父进程的pid.同时os.getpid()可获得当前进程的pid.
1 | import os |
结果:
1 | Process (876) start... |
Python自带的多进程模块
windows没有fork().可以通过python提供的通用多进程模块multiprocessing创建多进程。
创建多进程需要导入Process模块:
from multiprocess import Process
使用
p = Process(target=function, args=(parament,...)
创建子进程实例.其中target=传入子进程需执行的函数本身function,args传入函数需要的参数.参数数量不固定.
之后使用
p.start()
运行实例.要等待该子进程运行结束再运行之后的代码可以使用:
p.join()
以下是一个例子:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15 from multiprocessing import Process
import os
# 子进程要执行的代码
def run_proc(name):
print 'Run child process %s (%s)...' % (name, os.getpid())
if __name__=='__main__':
print 'Parent process %s.' % os.getpid()
p = Process(target=run_proc, args=('test',))
print 'Process will start.'
p.start()
p.join()
print 'Process end.'
结果:
1 | Parent process 928. |
对于需启动大量子进程的情况,可使用Pool模块:
from multiprocessing import Pool
使用:
p = Pool(number)
创建进程池.其中number为进程池包含子进程数量.不写默认为CPU核数.
使用:p.map(function,inter)p.map_async(function,inter)p.apply(function, args=(parament,...)p.apply_async(function, args=(parament,...)
其中之一同时运行子进程.其中map只支持单个参数.inter为可迭代对象.async为非阻塞方式运行，即不等结果出来主程序也会继续运行后面的代码。而apply输入则是一个list中的数量超过一个的参数。
之后需关闭进程池:
p.close()
同时,需等待所有子进程运行结束可使用:
p.join()
简单的例子：
1
2
3
4
5
6
7
8
9
10
11
from multiprocessing import Pool
import os
def Pri(x):
print x*x
print os.getpid()
p = Pool(5)
for i in range(10):
p.apply_async(Pri, [i])
p.close()
1
2
3
4
5
6
7
8
9
10
from multiprocessing import Pool
import os
def Pri(x):
print x*x
print os.getpid()
p = Pool(5)
p.map(Pri, range(10))
p.close()
另一个例子:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20 from multiprocessing import Pool
import os, time, random
def long_time_task(name):
print 'Run task %s (%s)...' % (name, os.getpid())
start = time.time()
time.sleep(random.random() * 3)
end = time.time()
print 'Task %s runs %0.2f seconds.' % (name, (end - start))
if __name__=='__main__':
print 'Parent process %s.' % os.getpid()
p = Pool()
for i in range(5):
p.apply_async(long_time_task, args=(i,))
print 'Waiting for all subprocesses done...'
p.close()
p.join()
print 'All subprocesses done.'
结果:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
Parent process 669.
Waiting for all subprocesses done...
Run task 0 (671)...
Run task 1 (672)...
Run task 2 (673)...
Run task 3 (674)...
Task 2 runs 0.14 seconds.
Run task 4 (673)...
Task 1 runs 0.27 seconds.
Task 3 runs 0.86 seconds.
Task 0 runs 1.41 seconds.
Task 4 runs 1.91 seconds.
All subprocesses done.
进程间通讯
不同进程间可以通过
Queue,Pipe来通信.Pipe用于两个进程间通信,Quene用于多个进程间通信.在只有两个进程通信的情况下Pipe效率高于Queue.- Pipe
导入
Pipe模块:from multiprocessing import Pipe创建Pipe通信的两端(返回一个双元素的list):
p = Pipe(duplex=False)
其中duplex=False表示该Pipe只能单向通信.默认不写该参数为双向通信.p[0],p[1]可以分别作为两个子进程的参数传递给子进程函数.也可以只传递一端给子进程,另一端交给父进程.Pipe的两端可通过p.send()传送值,p.recv()接收值.例子1:
1
2
3
4
5
6
7
8
9
10
11
12from multiprocessing import Process, Pipe
def f(conn):
conn.send([42, None, 'hello'])
conn.close()
if __name__ == '__main__':
parent_conn, child_conn = Pipe()
p = Process(target=f, args=(child_conn,))
p.start()
print parent_conn.recv() # prints "[42, None, 'hello']"
p.join()
例子2:
1 | import multiprocessing as mul |
- Queue
导入Queue模块:
from multiprocessing import Queue
创建Queue对象:
q = Queue(max)
其中max表示对象中可以存放的最大数量.
q可作为全局变量使用,也可以作为参数传递给子进程.
使用q.put()在Queue对象中放入需传递的值,q.get()取出值.
例子1:
1 | from multiprocessing import Process,Queue |
例子2:
1 | import multiprocessing |
多线程
多任务除了使用多进程外还可以使用多线程来完成.单个进程中可以有多个线程,它们共享进程中的数据.
python中可使用高级模块Threading来创建多线程.其使用方法与multiprocessing相似.
导入Threading模块:
import Threading
threading 模块提供的常用方法：
threading.currentThread(): 返回当前的线程变量。 (也可以使用threading.current_thread())
threading.enumerate(): 返回一个包含正在运行的线程的list。正在运行指线程启动后、结束前，不包括启动前和终止后的线程。
threading.activeCount(): 返回正在运行的线程数量，与len(threading.enumerate())有相同的结果。
threading模块提供的类：
Thread, Lock, Rlock, Condition, [Bounded]Semaphore, Event, Timer, local.
构建新线程实例:
t = Threading.thread(target=function,...)
同时构建实例支持以下几种方法:
Thread(group=None, target=None, name=None, args=(), kwargs={})
group: 线程组，目前还没有实现，库引用中提示必须是None；
target: 要执行的方法；
name: 线程名；
args/kwargs: 要传入方法的参数。
实例支持以下方法:
isAlive(): 返回线程是否在运行。正在运行指启动后、终止前。
get/setName(name): 获取/设置线程名。
is/setDaemon(bool): 获取/设置是否守护线程。初始值从创建该线程的线程继承。当没有非守护线程仍在运行时，程序将终止。
start(): 启动线程。
join([timeout]): 阻塞当前上下文环境的线程，直到调用此方法的线程终止或到达指定的timeout（可选参数）。
使用:
t.start()
运行新线程.
如需等待线程运行结束:
t.join()
锁
由于多线程共享进程中的变量, 如果直接使用多线程修改变量的话容易出问题.所以多线程中一般会创建锁.创建锁:
lock = threading.Lock()
此时有了一个锁的实例.锁的实例方法:
acquire([timeout]): 使线程进入同步阻塞状态，尝试获得锁定。
release(): 释放锁。使用前线程必须已获得锁定，否则将抛出异常。在每个线程需要修改变量前调用实例方法,尝试将修改变量的过程置于锁中,不让其他线程修改变量:
lock.acquire()修改之后需要释放锁:
lock.release()例子1:
1
2
3
4
5
6
7
8
9
10
11
12
13balance = 0
lock = threading.Lock()
def run_thread(n):
for i in range(100000):
# 先要获取锁:
lock.acquire()
try:
# 放心地改吧:
change_it(n)
finally:
# 改完了一定要释放锁:
lock.release()例子2:
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# encoding: UTF-8
import threading
import time
data = 0
lock = threading.Lock()
def func():
global data
print '%s acquire lock...' % threading.currentThread().getName()
# 调用acquire([timeout])时，线程将一直阻塞，
# 直到获得锁定或者直到timeout秒后（timeout参数可选）。
# 返回是否获得锁。
if lock.acquire():
print '%s get the lock.' % threading.currentThread().getName()
data += 1
time.sleep(2)
print '%s release lock...' % threading.currentThread().getName()
# 调用release()将释放锁。
lock.release()
t1 = threading.Thread(target=func)
t2 = threading.Thread(target=func)
t3 = threading.Thread(target=func)
t1.start()
t2.start()
t3.start()其他锁及线程间通信见参考资料5中.
多线程的全局变量与局部变量
多线程之间修改全局变量需要加锁. 在线程的函数中创建局部变量可以解决加锁问题, 但如果线程需要运行不同函数, 函数之间需要共享变量, 局部变量调用不是很方便.threading.local()可以解决这个问题.
localschool = threading.local()创建实例后, 不同线程在同时使用实例时不会产生冲突.
例子:
1 | import threading |
协程
协程是单线程在不同的函数间中断并相互切换的一种运行模式.比如在函数A运行遇到阻塞时转向运行函数B,等到函数B运行结束再回来接着运行函数A.与多线程相比协程没有锁的问题.协程可以在IO密集的程序中节省IO等待时间,提高运行效率.
yield
python的生成器yield一定程度上支持协程.定义生成器yield可以直接在函数定义中将return换成yield.在调用生成器函数时首先将生成器赋给变量,通过变量的.next()方法调用生成器生成第一个值.再次调用.next()方法可生成第二个值..send(value)方法可在调用生成器时给它传递一个参数.
例子:1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25import time
def consumer():
r = ''
while True:
n = yield r
if not n:
return
print('[CONSUMER] Consuming %s...' % n)
time.sleep(1)
r = '200 OK'
def produce(c):
c.next()
n = 0
while n < 5:
n = n + 1
print('[PRODUCER] Producing %s...' % n)
r = c.send(n)
print('[PRODUCER] Consumer return: %s' % r)
c.close()
if __name__=='__main__':
c = consumer()
produce(c)
结果:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
[PRODUCER] Producing 1...
[CONSUMER] Consuming 1...
[PRODUCER] Consumer return: 200 OK
[PRODUCER] Producing 2...
[CONSUMER] Consuming 2...
[PRODUCER] Consumer return: 200 OK
[PRODUCER] Producing 3...
[CONSUMER] Consuming 3...
[PRODUCER] Consumer return: 200 OK
[PRODUCER] Producing 4...
[CONSUMER] Consuming 4...
[PRODUCER] Consumer return: 200 OK
[PRODUCER] Producing 5...
[CONSUMER] Consuming 5...
[PRODUCER] Consumer return: 200 OK
更多关于`yield`见参考资料6.
gevent
gevent模块为python提供了完整的协程实现. 使用需先导入模块: `import gevent`在网络通信中一般还会导入
monkey模块以将默认socket替换为可协程的socket:1
2from gevent import monkey
monkey.patch_socket()或者将所有阻塞式调用,包括
socket,ssl,threading,select都替换为异步式:1
2from gevent import monkey
monkey.patch_all()这种替换调用一般放在第一行.替换后这些调用会自动处理阻塞问题.
在需使用协程时要用:
g = gevent.spawn(function,parament)
使用协程方式启动函数.参数为函数的参数.等待任务结束可以使用:
g.join()等待所有任务结束可以使用:
gevent.joinall(spawnlist)不过使用
monkey模块补丁自动处理有时候不能满足要求.这时我们可以使用其他模块.
如自动处理会并发所有连接,如果需要限制并发数量的话可以使用Pool模块.from gevent.pool import Pool新建一个
Pool池:p = Pool(number)
number为最高并发数在并发池中启动函数:
p.spawn(function,parament)等待所有任务结束:
p.join()需要直接指定跳转时用
sleep函数:gevent.sleep(time)
其中time表示此处至少要阻塞time秒.
参考资料: