标签：queue obj get 队列模型间通信 print 进程

（一）引入

（1）什么是进程间的通信 IPC

进程间通信（Inter-Process Communication, IPC）是指两个或多个进程之间进行信息交换的过程
它是一种计算机编程技术，用于在不同的进程之间共享数据和资源

（2）如何实现进程间通信

借助于消息队列，进程可以将消息放入队列中，然后由另一个进程从队列中取出
这种通信方式是非阻塞的，即发送进程不需要等待接收进程的响应即可继续执行
multiprocessing模块支持两种形式：队列和管道，这两种方式都是使用消息传递的

（3）什么是管道

管道是一种半双工的通信机制，即只能在一个方向上进行数据传输。
子进程可以通过继承父进程的管道来实现通信。
stdin、stdout和stderr是Python中的三个内置文件对象，它们分别代表标准输入、标准输出和标准错误。
这些对象也可以作为管道使用。
当我们在一个进程中使用read方法读取管道内的消息后，其他进程将无法再获取该管道内的任何其他消息。
因此，我们需要使用锁或其他同步机制来确保多个进程能够正确地访问和修改共享资源。

（4）什么是队列（管道+锁）

队列是一种线程安全的数据结构，它支持在多线程环境中高效的实现生产者-消费者模型
队列的特性是先进先出，即先插入队列的数据将先被取出
堆栈式一种先进后出的数据结构，与队列相反，最后插入的数据将首先被取出

（5）进程间通信的目的

存是为了更方便的取
千方百计的存
简单快捷的取

（二）队列介绍

（1）创建队列的类

底层就是以管道和锁定的方式实现

（1）语法

# 导入模块
import queue
q=queue.Queue()

创建共享的进程队列
Queue是多进程安全的队列
可以使用Queue实现多进程之间的数据传递

（2）参数

Queue（maxsize）
maxsize 是队列中允许存在的最大项数，可以指定项数，省略则表示无大小限制

（2）方法介绍

（1）主要方法

q.put() 插入数据到队列中
- 用以插入数据到队列中，put方法还有两个可选参数：blocked和timeout。
- 如果blocked为True（默认值），并且timeout为正值，该方法会阻塞timeout指定的时间，直到该队列有剩余的空间。如果超时，会抛出Queue.Full异常
- 如果blocked为False，但该Queue已满，会立即抛出Queue.Full异常。
q.get() 获取队列中的数据并将数据从队列中删除
- 可以从队列读取并且删除一个元素，同样，get方法有两个可选参数：blocked和timeout。
- 如果blocked为True（默认值），并且timeout为正值，那么在等待时间内没有取到任何元素，会抛出Queue.Empty异常。
- 如果blocked为False，有两种情况存在，如果Queue有一个值可用，则立即返回该值，否则，如果队列为空，则立即抛出Queue.Empty异常.
q.get_nowait()
- 同q.get(False)
q.put_nowait()
- 同g.put(False)
q.empty() 判断队列是否为空
- 调用此方法时q为空则返回True，该结果不可靠，比如在返回True的过程中，如果队列中又加入了项目。
q.full() 判断队列是否已满
- 调用此方法时q已满则返回True，该结果不可靠，比如在返回True的过程中，如果队列中的项目被取走。
- q.qsize() 返回队列中目前项目的正确数量
  - 返回队列中目前项目的正确数量，结果也不可靠，理由同q.empty()和q.full()一样

（2）了解方法

q.cancel_join_thread()
- 不会在进程退出时自动连接后台线程。可以防止join_thread()方法阻塞
q.close()
- 关闭队列，防止队列中加入更多数据。
- 调用此方法，后台线程将继续写入那些已经入队列但尚未写入的数据，但将在此方法完成时马上关闭。
- 如果q被垃圾收集，将调用此方法。
- 关闭队列不会在队列使用者中产生任何类型的数据结束信号或异常。
- 例如，如果某个使用者正在被阻塞在get()操作上，关闭生产者中的队列不会导致get()方法返回错误。
q.join_thread()
- 连接队列的后台线程。
- 此方法用于在调用q.close()方法之后，等待所有队列项被消耗。
- 默认情况下，此方法由不是q的原始创建者的所有进程调用。调用q.cancel_join_thread方法可以禁止这种行为

（3）方法示例

（1）q.get()

# 导入模块
import queue
# 创建队列对象
q=queue.Queue()

# 向队列中加入数据
q.put(1)

# 取出队列中的数据
print(q.get())# 1

# 如果队列中无数据 不会报错但会卡住
print(q.get())

"""
	当队列中的数据已经满了之后，还要放入数据那么程序就会阻塞，指定队列中有位置让出来（不会报错）
	当队列中的数据已经全部取出后，仍要取出数据 程序也会阻塞
"""

(2)q.get_nowait() ：队列无数据时会报错

# 导入模块
import queue
# 创建队列对象
q=queue.Queue()

# 向队列中加入数据
q.put(1)

# 取出队列中的数据
print(q.get())# 1

# 如果队列中无数据 不会报错但会卡住
# print(q.get())

# 当队列中的数据已被取空后 q.get_nowait()会报错
print(q.get_nowait())# _queue.Empty

（3）q.get(timeout=3):设置延迟时间，推迟指定时间后报错

# 导入模块
import queue

# 创建队列对象
q=queue.Queue()

# 向队列中放入数据
q.put(1)
q.put(2)
# 取出队列中的数据
print(q.get())
print(q.get(timeout=3))

# 设置延时时间,推迟指定 s 数  报错
print(q.get(timeout=3))
#     raise Empty _queue.Empty

（4）q.full() :判断队列是否已经填满

# 导入模块
import queue
# 创建队列对象
q=queue.Queue(3)
q.put(1)
q.put(2)
# 创建队列对象,设置队列中最多有三个元素,放入两个时,判断队列是否已经填满了
print(q.full())# False
q.put(3)
# 设置队列中最多有三个元素,向队列中放入三个元素后,判断队列是否已经填满了
print(q.full())# True

（5）q.empty():判断当前队列是否为空

# 导入模块
import queue

# 创建队列对象
q=queue.Queue(3)
# 此时不向队列中,添加任何元素时,判断当前队列是否为空
print(q.empty())# True
# 向队列中放入元素,判断队列是否为空
q.put(1)
print(q.empty())# False

（4）队列queue模块小结

"""queue模块"""

# 导入模块
import queue

# 创建队列对象
# q=queue.Queue(maxsize=5)
# 可以指定参数,指定的参数指的时队列中的元素的个数,
# 也可以不设置参数,那么元素的个数可以是随意的
q=queue.Queue(maxsize=5)

# q.put()方法  向队列中添加元素
q.put(1)
q.put(2)

# q.get()方法,取出队列中的元素
print(q.get()) # 1
print(q.get()) # 2
# 如果想要取得元素个数 大于队列中得元素个数 q.get()方法会发送阻塞,但不会报错
# print(q.get())

# q.get_nowait()方法
# 当队列中没有元素数据了,q.get_nowait()方法,不会发送阻塞,会报错
q.get_nowait()
#     raise Empty
# _queue.Empty

# q.full()方法:判断队列是否已经填满
# 满了就返回True,没有满就返回False
q.full()

# q.empty()方法:判断队列是否为空
# 空了就返回True,没空就返回False
q.empty()

以上方法在多进程的情况下不精确
多进程的队列切换时间是非常短的，存在可能判断失误

（三）进程间通信（IPC机制）

（1）什么是IPC机制

IPC机制指进程间通信机制（Inter-Process Communication），它是指在不同进程间传输数据或者信息的一种机制。
在多进程操作系统中，各个进程相互独立，不能直接访问对方的内存空间，所以必须通过特殊的通信方式实现进程之间的信息交换和协调。
常见的IPC机制包括管道、消息队列、共享内存和信号量等方式。
其中，管道、消息队列和共享内存都是用于进程之间的数据传输
而信号量则是用于进程之间的同步与互斥。
通过使用IPC机制，可以使得不同进程能够有效地协同工作，在复杂的程序设计中起到十分重要的作用。

（2）子进程与主进程之间通过队列进行通信

主进程与子进程之前借助于队列进行通信

"""主进程与子进程间的通信"""
# 子进程向队列中填入数据，主进程获取出队列着数据
import time
# 导入模块
from multiprocessing import Process,Queue
def test(q):
    print(f"子进程")
    q.put(11)
    time.sleep(2)
if __name__ == '__main__':
    print("主进程开始")
    # 创建队列对象
    q=Queue()
    # 创建进程对象
    p=Process(target=test,args=(q,))
    p.start()
    p.join()
    print(q.get())
    print("主进程结束")
    
    
# 主进程向队列中添加数据，子进程获取出队列中的数据
"""主进程添加数据,子进程获取数据"""
# 导入模块
from multiprocessing import Process,Queue

def test(q):
    msg=q.get()
    print(f"这是子进程获取到的列表元素{msg}")

def main():
    print("主进程开始")
    # 创建队列对象
    q=Queue()
    # 主进程向列表中添加元素
    q.put("主进程添加进入的")
    # 创建进程对象
    p=Process(target=test,args=(q,))
    # 启动进程对象
    p.start()
    p.join()
    print("主进程结束")
if __name__ == '__main__':
    main()

（3）子进程与子进程之间借助队列进行通信

子进程与子进程之间借助队列进行通信

"""子进程与子进程之间借助队列进行通信"""
# 导入模块
from multiprocessing import Process,Queue

# 向队列中添加元素
def test1(q):
    q.put(1)
    print("子进程向队列中添加元素成功!!")

# 获取出队列中的元素
def test2(q):
    msg=q.get()
    print(f"子进程获取出队列中的元素{msg}!!")

def main():
    # 创建队列对象
    q=Queue()
    # 创建进程对象
    p1=Process(target=test1,args=(q,))
    p2=Process(target=test2,args=(q,))
    p1.start()
    p2.start()

if __name__ == '__main__':
    main()
# 子进程向队列中添加元素成功!!
# 子进程获取出队列中的元素1!!




"""子进程与子进程之间借助队列进行通信"""

# 导入模块
from multiprocessing import  Process,Queue
# 创建一个子进程函数
def producer(queue_obj):
    # 生产者进程向队列中添加数据
    queue_obj.put(1111)
    print("生产者进程已添加数据!!!")
# 再创建一个子进程函数
def customer(queue_obj):
    # 消费者对象获取出队列的数据
    msg=queue_obj.get()
    print(f"消费者进程获取出队列中的{msg}数据")
# 创建一个主函数
def main():
    # 创建队列对象
    queue_obj=Queue()
    # 创建进程对象
    p1=Process(target=producer,args=(queue_obj,))
    p2=Process(target=customer,args=(queue_obj,))
    p1.start()
    p2.start()
    p1.join()
    p2.join()

if __name__ == '__main__':
    main()

# 生产者进程已添加数据!!!
# 消费者进程获取出队列中的1111数据

（四）生产者和消费者模型

（1）理论

（1）生产者模型

生产者模型和消费者模型是指通过利用队列解耦生产者和消费者的一种并发编程模型。
在生产者模型中，生产者负责将数据放入共享队列中，而消费者则从队列中取出数据进行处理。
生产者和消费者之间通过共享这个队列来进行信息的交流。
这种模型适用于生产者和消费者之间的处理速度不一致的情况，同时还能够保证数据传输的安全性和正确性。

（2）消费者模型

在消费者模型中，消费者负责向队列中插入任务，而由线程池中的工作线程进行任务的处理。
消费者和工作线程之间通过共享线程池中的任务队列来完成任务分发和执行。
这种模型适用于任务处理需要一定时间的情况，能够充分利用多线程的优势提高系统的并发性能，提高效率。

（3）小结

生产者：生产/制造东西
消费者：消费/处理东西
该模型还需要一个媒介

（2）实现

消费者大于生产者
在消费者获取初数据前判断队列李是否为空
不为空取出数据
为空结束消费

# 导入模块
from multiprocessing import Process,Queue
import time

# 创建生产者函数
def producer(queue_obj):
    # 生产者---向列表中添加数据
    for i in range(1,3):
        queue_obj.put(i)
        # time.sleep(2)
    # 添加标志位
    # queue_obj.put('q')

# 创建消费者函数
def customer(queue_obj):
    while True:
        if  not queue_obj.empty():
            msg=queue_obj.get()
            # if msg=='q':
            #     print("消费者消费完毕!!")
            #     break
            print(f"消费者取出队列中的数据---{msg}")
        else:
            break
# 创建主函数
def main():
    print("主进程开始")
    # 创建队列对象
    queue_obj=Queue()
    # 创建一个空的进程列表
    p_list=[]
    # 创建生产者对象
    p1=Process(target=producer,args=(queue_obj,))
    p2=Process(target=producer,args=(queue_obj,))

    # 创建消费者对象
    c1=Process(target=customer,args=(queue_obj,))
    c2=Process(target=customer,args=(queue_obj,))
    # 启动进程对象
    p1.start()
    p2.start()
    c1.start()
    c2.start()
    p_list.append(p1)
    p_list.append(p2)
    p_list.append(c1)
    p_list.append(c2)
    for task in p_list:
        task.join()
    print("主进程结束")
if __name__ == '__main__':
    main()


# 主进程开始
# 消费者取出队列中的数据---1
# 消费者取出队列中的数据---2
# 消费者取出队列中的数据---1
# 消费者取出队列中的数据---2
# 主进程结束

给生产者加一个标准位
消费者获取出队列中的标准位时
结束消费

# 导入模块
from multiprocessing import Process,Queue
import time

# 创建生产者函数
def producer(queue_obj):
    # 生产者---向列表中添加数据
    for i in range(1,3):
        queue_obj.put(i)
        # time.sleep(2)
    # 添加标志位
    queue_obj.put('q')

# 创建消费者函数
def customer(queue_obj):
    while True:
        msg=queue_obj.get()
        if msg=='q':
            print("消费者消费完毕!!")
            break
        print(f"消费者取出队列中的数据---{msg}")

# 创建主函数
def main():
    print("主进程开始")
    # 创建队列对象
    queue_obj=Queue()
    # 创建一个空的进程列表
    p_list=[]
    # 创建生产者对象
    p1=Process(target=producer,args=(queue_obj,))
    p2=Process(target=producer,args=(queue_obj,))

    # 创建消费者对象
    c1=Process(target=customer,args=(queue_obj,))
    c2=Process(target=customer,args=(queue_obj,))
    # 启动进程对象
    p1.start()
    p2.start()
    c1.start()
    c2.start()
    p_list.append(p1)
    p_list.append(p2)
    p_list.append(c1)
    p_list.append(c2)
    for task in p_list:
        task.join()
    print("主进程结束")
if __name__ == '__main__':
    main()

JoinableQueue模块
每当向队列对象中存入数据的时候，队列对象内部会有一个计数器+1
每当调用异常task——done（）方法的时候，队列对象内部的计数器-1
join（）当计数器为0时，继续执行代码

# 导入模块
from multiprocessing import Process,Queue,JoinableQueue
import time
# 生产者
def producer(queue_obj):
    for i in range(1,5):
        queue_obj.put(i)
    queue_obj.join()
# 消费者
def customer(queue_obj):
    while True:
        msg=queue_obj.get()
        print(f"消费者取出队列中的数据{msg}")
        # queue_obj.task_done() 会自动检索队列中的数据个数,如果数据大于1就忽略,如果数据等于0就自动结束
        queue_obj.task_done()
# 主进程
def main():
    # 创建队列对象
    queue_obj=JoinableQueue()
    # 创建空的进程列表
    task_list=[]

    # 创建进程对象
    # 生产者对象
    p1=Process(target=producer,args=(queue_obj,))
    p2=Process(target=producer,args=(queue_obj,))
    # 消费者对象
    c1 = Process(target=customer, args=(queue_obj,))
    c2 = Process(target=customer, args=(queue_obj,))
    # 把消费者设置为守护进程
    # 随着主进程的结束而结束
    c1.daemon=True
    c2.daemon=True
    
    task_list.append(p1)
    task_list.append(p2)
    task_list.append(c1)
    task_list.append(c2)
    for task in task_list:
        task.start()
    # 主进程等待 生产者
    p1.join()
    p2.join()
    print("主进程结束")
if __name__ == '__main__':
    main()

# 消费者取出队列中的数据1
# 消费者取出队列中的数据2
# 消费者取出队列中的数据3
# 消费者取出队列中的数据4
# 消费者取出队列中的数据1
# 消费者取出队列中的数据2
# 消费者取出队列中的数据3
# 消费者取出队列中的数据4
# 主进程结束

生产者和消费者模型小结：

生产者和消费者模型总结：
- 生产者：生产数据生产完所有数据以后要加一个标志位
- 消费者：消费数据从生产者生产的数据中拿出所有数据
解决生产者和消费者不平稳一：
- Queue
  - 在生产者生产数据的末尾加一个标志位
  - 消费者消费数据，消费者消费到最后一个数据得到生产者加入的标准位时，结束
解决生产者和消费者不平稳二：
- JoinableQueue:
  - 在生产者生产数据的时候放一个标志位---队列对象.join()
  - 在消费者消费数据的时候---正常消费---在消费数据的结尾加入队列对象.task_done() ---自动检索生产者是否存在join()
  - 自动检索队列中的数据个数---大于1继续继续，等于0结束

(五)JoinableQueue模块

（1）介绍

JoinableQueue 模块：这就像时一个Queue对象，但队列允许项目的使用者通知生产者项目已经被成功处理
通知进程时使用共享的信号和条件变量来实现的

（2）使用

JoinableQueue([maxsize])
- maxsize是队列中允许最大项数，省略则无大小限制。

from multiprocessing import JoinableQueue
    
q = JoinableQueue()

# 方法介绍：JoinableQueue的实例p除了与Queue对象相同的方法之外还具有：

# q.task_done()：使用者使用此方法发出信号，表示q.get()的返回项目已经被处理。如果调用此方法的次数大于从队列中删除项目的数量，将引发ValueError异常

# q.join():生产者调用此方法进行阻塞，直到队列中所有的项目均被处理。阻塞将持续到队列中的每个项目均调用q.task_done（）方法为止

标签：queue,obj,get,队列,模型,间通信,print,进程
From： https://www.cnblogs.com/suyihang/p/17987124

进程间通信（队列和生产消费模型）

（一）引入

（1）什么是进程间的通信 IPC

（2）如何实现进程间通信

（3）什么是管道

（4）什么是队列（管道+锁）

（5）进程间通信的目的

（二）队列介绍

（1）创建队列的类

（1）语法

（2）参数

（2）方法介绍

（1）主要方法

（2）了解方法

（3）方法示例

（1）q.get()

(2)q.get_nowait() ：队列无数据时会报错

（3）q.get(timeout=3):设置延迟时间，推迟指定时间后报错

（4）q.full() :判断队列是否已经填满

（5）q.empty():判断当前队列是否为空

（4）队列queue模块小结

（三）进程间通信（IPC机制）

（1）什么是IPC机制

（2）子进程与主进程之间通过队列进行通信

（3）子进程与子进程之间借助队列进行通信

（四）生产者和消费者模型

（1）理论

（1）生产者模型

（2）消费者模型

（3）小结

（2）实现

生产者和消费者模型小结：

(五)JoinableQueue模块

（1）介绍

（2）使用

相关文章

赞助商

阅读排行

进程间通信（队列和生产消费模型）

（一）引入

（1）什么是进程间的通信 IPC

（2）如何实现进程间通信

（3）什么是管道

（4）什么是队列（管道+锁）

（5）进程间通信的目的

（二）队列介绍

（1）创建队列的类

（1）语法

（2）参数

（2）方法介绍

（1）主要方法

（2）了解方法

（3）方法示例

（1）q.get()

(2)q.get_nowait() ： 队列无数据时会报错

（3）q.get(timeout=3):设置延迟时间，推迟指定时间后报错

（4）q.full() :判断队列是否已经填满

（5）q.empty():判断当前队列是否为空

（4）队列queue模块小结

（三）进程间通信（IPC机制）

（1）什么是IPC机制

（2）子进程与主进程之间通过队列进行通信

（3）子进程与子进程之间借助队列进行通信

（四）生产者和消费者模型

（1）理论

（1）生产者模型

（2）消费者模型

（3）小结

（2）实现

生产者和消费者模型小结：

(五)JoinableQueue模块

（1）介绍

（2）使用

相关文章

赞助商

阅读排行

(2)q.get_nowait() ：队列无数据时会报错