这篇文章主要介绍Python并发编程的示例分析,文中介绍的非常详细,具有一定的参考价值,感兴趣的小伙伴们一定要看完!
0x01 multipleprocessing
与使用线程的 threading 模块类似, multipleprocessing
模块提供许多高级 API 。最常见的是 Pool 对象了,使用它的接口能很方便地写出并发执行的代码。
from multiprocessing import Pool
def f(x):
return x * x
if __name__ == '__main__':
with Pool(5) as p:
# map方法的作用是将f()方法并发地映射到列表中的每个元素
print(p.map(f, [1, 2, 3]))
# 执行结果
# [1, 4, 9]
关于 Pool 下文中还会提到,这里我们先来看 Process 。
Process
要创建一个进程可以使用 Process 类,使用 start() 方法启动进程。
from multiprocessing import Process
import os
def echo(text):
# 父进程ID
print("Process Parent ID : ", os.getppid())
# 进程ID
print("Process PID : ", os.getpid())
print('echo : ', text)
if __name__ == '__main__':
p = Process(target=echo, args=('hello process',))
p.start()
p.join()
# 执行结果
# Process Parent ID : 27382
# Process PID : 27383
# echo : hello process
进程池
正如开篇提到的 multiprocessing
模块提供了 Pool 类可以很方便地实现一些简单多进程场景。 它主要有以下接口
apply(func[, args[, kwds]])
执行 func(args,kwds) 方法,在方法结束返回前会阻塞。
apply_async(func[, args[, kwds[, callback[, error_callback]]]])
异步执行 func(args,kwds) ,会立即返回一个 result 对象,如果指定了 callback 参数,结果会通过回调方法返回,还可以指定执行出错的回调方法 error_callback()
map(func, iterable[, chunksize])
类似内置函数 map() ,可以并发执行 func ,是同步方法
map_async(func, iterable[, chunksize[, callback[, error_callback]]])
异步版本的 map
close()
关闭进程池。当池中的所有工作进程都执行完毕时,进程会退出。
terminate()
终止进程池
join()
等待工作进程执行完,必需先调用 close() 或者 terminate()
from multiprocessing import Pool
def f(x):
return x * x
if __name__ == '__main__':
with Pool(5) as p:
# map方法的作用是将f()方法并发地映射到列表中的每个元素
a = p.map(f, [1, 2, 3])
print(a)
# 异步执行map
b = p.map_async(f, [3, 5, 7, 11])
# b 是一个result对象,代表方法的执行结果
print(b)
# 为了拿到结果,使用join方法等待池中工作进程退出
p.close()
# 调用join方法前,需先执行close或terminate方法
p.join()
# 获取执行结果
print(b.get())
# 执行结果
# [1, 4, 9]
# <multiprocessing.pool.MapResult object at 0x10631b710>
# [9, 25, 49, 121]
map_async() 和 apply_async() 执行后会返回一个 class multiprocessing.pool.AsyncResult 对象,通过它的 get() 可以获取到执行结果, ready() 可以判断 AsyncResult 的结果是否准备好。
进程间数据的传输
multiprocessing 模块提供了两种方式用于进程间的数据共享:队列( Queue )和管道( Pipe )
Queue 是线程安全,也是进程安全的。使用 Queue 可以实现进程间的数据共享,例如下面的 demo 中子进程 put 一个对象,在主进程中就能 get 到这个对象。 任何可以序列化的对象都可以通过 Queue 来传输。
from multiprocessing import Process, Queue
def f(q):
q.put([42, None, 'hello'])
if __name__ == '__main__':
# 使用Queue进行数据通信
q = Queue()
p = Process(target=f, args=(q,))
p.start()
# 主进程取得子进程中的数据
print(q.get()) # prints "[42, None, 'hello']"
p.join()
# 执行结果
# [42, None, 'hello']
Pipe() 返回一对通过管道连接的 Connection 对象。这两个对象可以理解为管道的两端,它们通过 send() 和 recv() 发送和接收数据。
from multiprocessing import Process, Pipe
def write(conn):
# 子进程中发送一个对象
conn.send([42, None, 'hello'])
conn.close()
def read(conn):
# 在读的进程中通过recv接收对象
data = conn.recv()
print(data)
if __name__ == '__main__':
# Pipe()方法返回一对连接对象
w_conn, r_conn = Pipe()
wp = Process(target=write, args=(w_conn,))
rp = Process(target=read, args=(r_conn,))
wp.start()
rp.start()
# 执行结果
# [42, None, 'hello']
需要注意的是,两个进程不能同时对一个连接对象进行 send 或 recv 操作。
同步
我们知道线程间的同步是通过锁机制来实现的,进程也一样。
from multiprocessing import Process, Lock
import time
def print_with_lock(l, i):
l.acquire()
try:
time.sleep(1)
print('hello world', i)
finally:
l.release()
def print_without_lock(i):
time.sleep(1)
print('hello world', i)
if __name__ == '__main__':
lock = Lock()
# 先执行有锁的
for num in range(5):
Process(target=print_with_lock, args=(lock, num)).start()
# 再执行无锁的
# for num in range(5):
# Process(target=print_without_lock, args=(num,)).start()
有锁的代码将每秒依次打印
hello world 0
hello world 1
hello world 2
hello world 3
hello world 4
如果执行无锁的代码,则在我的电脑上执行结果是这样的
hello worldhello world 0
1
hello world 2
hello world 3
hello world 4
除了 Lock ,还包括 RLock 、 Condition 、 Semaphore 和 Event 等进程间的同步原语。其用法也与线程间的同步原语很类似。 API 使用可以参考文末中引用的文档链接。
在工程中实现进程间的数据共享应当优先使用 队列或管道。
以上是“Python并发编程的示例分析”这篇文章的所有内容,感谢各位的阅读!希望分享的内容对大家有帮助,更多相关知识,欢迎关注亿速云行业资讯频道!
亿速云「云服务器」,即开即用、新一代英特尔至强铂金CPU、三副本存储NVMe SSD云盘,价格低至29元/月。点击查看>>
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。