前端网站开发一个月多少钱,seo搜索引擎优化招聘,网站开发程序制作域名备案,免费网站申请域名Python中进程的Queue 和进程锁,以及进程池的创建 文章目录 Python中进程的Queue 和进程锁,以及进程池的创建前言一、进程间同步通信(Queue)二、进程锁#xff08;Lock#xff09;三、创建进程池Poorpool 类方法: End! 前言
大家好,上一篇文章,我们初步接触了进程的概念及其应…Python中进程的Queue 和进程锁,以及进程池的创建 文章目录 Python中进程的Queue 和进程锁,以及进程池的创建前言一、进程间同步通信(Queue)二、进程锁Lock三、创建进程池Poorpool 类方法: End! 前言
大家好,上一篇文章,我们初步接触了进程的概念及其应该如何创建一个进程等等,今天我们继续来深入学习一下进程中的进程间同步 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 Queueq Queue(3) # 创建一个Queue 对象,设置最大可接受put对象为3个
q.put(消息1)
q.put(消息2)
print(q.full()) # 判断队列是否满了 这里面输出False
q.put(消息3)
print(q.full()) #判断队列是否满了 这里面输出Trueprint(q.get()) # 获取队列中的消息,获取后把消息从队列中删除if not q.full(): # 判断队列是否满,如果不满就继续将消息写入队列q.put(消息4)print(q.get())
print(q.get())
print(q.get())运行结果: from multiprocessing import Queueq Queue(3) # 创建一个Queue 对象,设置最大可接受put对象为3个
q.put(消息1)
q.put(消息2)
print(q.full()) # 判断队列是否满了 这里面输出False
q.put(消息3)
print(q.full()) #判断队列是否满了 这里面输出Trueprint(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 randomdef 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:breakif __name__ __main__:q multiprocessing.Queue() # 创建一个Queue对象 qpw multiprocessing.Process(targetwrite, args(q, )) # 创建一个子进程 pw ,并将queue 对象传入pr multiprocessing.Process(targetread, 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, Processdef 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(targetfunc, 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, Managerdef 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(targetfunc, 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 timedef run(f):time.sleep(1)return f * fif __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, randomdef 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!
讲的不好,多多见谅,我们下次再见!
更多优质文章点这里
