zoukankan      html  css  js  c++  java
  • IPC机制和生产者消费者模型

    IPC机制:(解决进程间的数据隔离问题)

      进程间通信:IPC(inter-Process Comminication)

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

      底层实现:管道+锁的方式实现

    # Queue([maxsize]) 
    
    创建共享的进程队列。maxsize是队列中允许的最大项数。如果省略此参数,则无大小限制。底层队列使用管道和锁定实现。另外,还需要运行支持线程以便队列中的数据传输到底层管道中。 
    Queue的实例q具有以下方法:
    
    q.get( [ block [ ,timeout ] ] ) 
    返回q中的一个项目。如果q为空,此方法将阻塞,直到队列中有项目可用为止。block用于控制阻塞行为,默认为True. 如果设置为False,将引发Queue.Empty异常(定义在Queue模块中)。timeout是可选超时时间,用在阻塞模式中。如果在制定的时间间隔内没有项目变为可用,将引发Queue.Empty异常。
    
    q.get_nowait( ) 
    同q.get(False)方法。
    
    q.put(item [, block [,timeout ] ] ) 
    将item放入队列。如果队列已满,此方法将阻塞至有空间可用为止。block控制阻塞行为,默认为True。如果设置为False,将引发Queue.Empty异常(定义在Queue库模块中)。timeout指定在阻塞模式中等待可用空间的时间长短。超时后将引发Queue.Full异常。
    
    q.qsize() 
    返回队列中目前项目的正确数量。此函数的结果并不可靠,因为在返回结果和在稍后程序中使用结果之间,队列中可能添加或删除了项目。在某些系统上,此方法可能引发NotImplementedError异常。
    
    
    q.empty() 
    如果调用此方法时 q为空,返回True。如果其他进程或线程正在往队列中添加项目,结果是不可靠的。也就是说,在返回和使用结果之间,队列中可能已经加入新的项目。
    
    q.full() 
    如果q已满,返回为True. 由于线程的存在,结果也可能是不可靠的(参考q.empty()方法)。

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

     二、消费者生产者模型实现

    # -*- coding: utf-8 -*-
    
    """
    生产者消费模型:
        生产者:生产数据
        消费者:处理数据
    
        解决供需不平衡问题
            定义一个队列,用来存放固定数量的数据
            解决一个生产者与消费者不需要直接打交道,两者通过队列进行数据传输
    """
    from multiprocessing import Process, JoinableQueue
    import time
    import random
    
    
    def producer(name, food, q):
        for i in range(1, 5):
            res = '%s%s' % (food, i)
            time.sleep(random.randint(1, 2))
            print('%s 生产了%s' % (name, res))
            q.put(res)  # 将生产的数据放入队列中
    
    
    def consumer(name, q):
        while True:
            res = q.get()
            if res is None:
                break
            time.sleep(random.randint(1, 2))
            print('消费者%s正在吃%s' % (name, res))
            q.task_done()  # 告诉你的队列,你已经将数据取出并且处理完毕
    
    
    if __name__ == '__main__':
        q = JoinableQueue()  # 生成一个队列对象
        p1 = Process(target=producer, args=('祝志伟', '包子', q))
        p2 = Process(target=producer, args=('莫福进', '饺子', q))
        p3 = Process(target=producer, args=('张金', '粽子', q))
    
        c1 = Process(target=consumer, args=('QZK', q))
        c2 = Process(target=consumer, args=('RAIN', q))
        p1.start()
        p2.start()
        p3.start()
        c1.daemon = True
        c2.daemon = True
        c1.start()
        c2.start()
        p1.join()
        p2.join()
        p3.join()
    
        q.join()  # 等待队列中数据全部取出
        print('主进程')
    

      

  • 相关阅读:
    Java学习之路----计算圆形的面积和周长
    数据库系统的基本组成内容
    软件测试的含义以及测试的对象
    wg sync.WaitGroup执行顺序
    go channel
    字符串操作
    scanf
    py停止工作
    jira索引失败
    py kafka
  • 原文地址:https://www.cnblogs.com/qianzhengkai/p/10826555.html
Copyright © 2011-2022 走看看