Python中进程的Queue 和进程锁,以及进程池的创建
前言
大家好,上一篇文章,我们初步接触了进程的概念及其应该如何创建一个进程等等,今天我们继续来深入学习一下进程中的进程间同步 Queue 和进程锁lock 还有进行创建一个进程池pool 等等,我们开始今天的学习吧 !
一、进程间同步通信(Queue)
我们在使用进程的时候,有时候我们需要进行进程之间的通信,multiprocessing 模块提供了Queue类 来实现不同进程之间的通信.
Queue():括号内传的参数表示最大可接受的消息数量;没有传参或者数量为负数,则表示最大可接受的数量没有上限(直到内存的尽头)
Queue.put(item, block, timeout) : 将item消息写入队列中,block默认为True.
- 如果block使用默认值,且没有设置timeout秒数,消息队列如果已经没有空间可以写入,将会被阻塞,直到从消息队列腾出空间为止
- 如果block使用默认值,设置了timeout秒数,则会等待timeout秒,若还没有空间,则会抛出异常’queue.full’.
我们首先先创建一个进程对象:
上图就是我们queue 中的方法,接下来我们看看如何使用queue.
from multiprocessing import Queue
q = Queue(3) # 创建一个Queue 对象,设置最大可接受put对象为3个
q.put('消息1')
q.put('消息2')
print(q.full()) # 判断队列是否满了 这里面输出False
q.put('消息3')
print(q.full()) #判断队列是否满了 这里面输出True
print(q.get()) # 获取队列中的消息,获取后把消息从队列中删除
if not q.full(): # 判断队列是否满,如果不满就继续将消息写入队列
q.put('消息4')
print(q.get())
print(q.get())
print(q.get())
运行结果:
from multiprocessing import Queue
q = Queue(3) # 创建一个Queue 对象,设置最大可接受put对象为3个
q.put('消息1')
q.put('消息2')
print(q.full()) # 判断队列是否满了 这里面输出False
q.put('消息3')
print(q.full()) #判断队列是否满了 这里面输出True
print(q.get())
if not q.full():
q.put('消息4')
# print(q.get())
# print(q.get())
# print(q.get())
if not q.empty(): # 判断队列是否为空
for i in range(q.qsize()):
print(q.get_nowait())
运行结果:
注意:
Queue.put_nowait(item)===>Queue.put(item, False)
Queue.get_nowait()===>Queue.get(False)
Queue.get(block, timeout):获取队列中的一条信息,然后将其从队列中移除,block默认为True
如果block使用默认值,且没有设置timeout秒数,消息队列如果为空,将会被阻塞,直到从消息队列读到消息为止.
如果block使用默认值,设置了timeout秒数,则会等待timeout秒,若还没有消息,则会抛出异常’queue.empty’.
如果block值为False,消息队列如果为空,则立即抛出异常’queue.empty
接下来我们进行一次完整的进程间的通信
import multiprocessing #导包
import time
import random
def write(q): # 定义一个函数
for i in ['a', 'b', 'c']:
print('放入:', i)
q.put(i) # 分别将a,b,c 消息传入队列
time.sleep(random.random())
def read(q):
while True:
if not q.empty():
print('获取:', q.get()) 获取队列中的消息
time.sleep(random.random())
else:
break
if __name__ == '__main__':
q = multiprocessing.Queue() # 创建一个Queue对象 q
pw = multiprocessing.Process(target=write, args=(q, )) # 创建一个子进程 pw ,并将queue 对象传入
pr = multiprocessing.Process(target=read, args=(q, )) # 创建一个子进程 pr,并将queue对象传入
pw.start() # pw 子进程开始
pw.join() # 等待 pw进程结束
pr.start() # pr 子进程开始
pr.join() # 等待 pr进程结束
print('结束')
运行结果:
二、进程锁（Lock）
Python进程锁是用来在多进程程序中对共享资源进行同步访问的一种机制。
当多个进程需要同时访问某个共享资源时，可能会出现竞争条件（race condition），导致数据不一致或错误的结果。进程锁可以用来保证在任意时刻只有一个进程能够访问共享资源，从而避免竞争条件的发生。
进程锁的主要作用是确保在某个进程正在使用共享资源时，其他进程无法访问该资源，直到当前进程释放锁为止。这样可以保证共享资源在任意时刻只会被一个进程修改，从而避免数据不一致或错误的结果。
Python中的进程锁可以通过使用multiprocessing模块中的Lock类来实现。通过调用lock.acquire()方法可以获得锁，其他进程如果尝试获得同一个锁则会被阻塞。而通过调用lock.release()方法可以释放锁，允许其他进程获得锁并访问共享资源。
需要注意的是，进程锁只能在同一个计算机的多个进程之间起作用，无法在不同计算机之间的进程之间起作用。如果需要进行跨网络的进程同步，可以考虑使用分布式锁的机制。
Python中可以使用多种方式实现进程锁，以下是常用的两种：
使用multiprocessing.Lock()：multiprocessing.Lock()是Python标准库中的一个进程锁实现。可以通过acquire()方法获得锁，并在任务完成后使用release()方法释放锁。
例如:
from multiprocessing import Lock, Process
def func(lock, num):
lock.acquire()
try:
# 临界区代码
print(f"进程{num}获得了锁")
finally:
lock.release()
print(f"进程{num}释放了锁")
if __name__ == "__main__":
lock = Lock()
processes = []
for i in range(5):
p = Process(target=func, args=(lock, i))
processes.append(p)
p.start()
for p in processes:
p.join()
也可使用multiprocessing.Manager().Lock()：multiprocessing.Manager()是一个多进程共享数据的管理器。可以通过Manager().Lock()方法创建一个进程锁。具体使用方式与multiprocessing.Lock()相同，只是创建锁的方式有所不同。
例如:
from multiprocessing import Process, Manager
def func(lock, num):
lock.acquire()
try:
# 临界区代码
print(f"进程{num}获得了锁")
finally:
lock.release()
print(f"进程{num}释放了锁")
if __name__ == "__main__":
manager = Manager()
lock = manager.Lock()
processes = []
for i in range(5):
p = Process(target=func, args=(lock, i))
processes.append(p)
p.start()
for p in processes:
p.join()
三、创建进程池Poor
Python进程池:是一种用于管理和复用多个进程的工具。进程池可以提高程序的并发处理能力，通过复用已经创建的进程，避免频繁创建和销毁进程的开销。
pool 类方法:
Pool()括号内表示处理的进程数量
apply_async(func, args, kwargs):使用非阻塞方式调用func(并行执行,阻塞就是必须等待上一个进程退出才能执行下一个进程)
close():关闭进程池,不在接收新的进程
terminate():不管任务是否完成,立即终止
join():阻塞,等待所有子进程结束,必须在close之后使用.
import multiprocessing
import time
def run(f):
time.sleep(1)
return f * f
if __name__ == '__main__':
test = [1, 2, 3, 4, 5, 6]
# print('顺序:')
# s = time.time() # 计算当前时间
# for i in test:
# print(run(i))
e = time.time()
# print('顺序执行时间:', int(e-s))
# map(函数名, 可迭代对象):循环将可迭代对象传递给函数执行
print('并发:')
pool = multiprocessing.Pool(5) # 创建能够有5条进程的进程池
r1 = pool.map(run, test)
pool.close()
pool.join()
e2 = time.time()
print('并发时间:', int(e2 - e))
print(r1)
运行结果:
在进程池里进行进程间的通信
例如：
import multiprocessing, os, time, random
def write(q):
print('write(%s)启动' % os.getpid())
for i in ['a', 'b', 'c', 'd', 'e']:
q.put(i)
def read(q):
print('read(%s)启动' % os.getpid())
for i in range(q.qsize()):
print('获取:', q.get())
if __name__ == '__main__':
print('主进程(%s)开始' % os.getpid())
q = multiprocessing.Manager().Queue()
pool = multiprocessing.Pool()
pool.apply_async(write, (q, ))
pool.apply_async(read, (q, ))
pool.close()
pool.join()
print('主进程结束')
运行结果：
End!
讲的不好,多多见谅,我们下次再见!
