【python中级】多进程之间通过Queue队列(先进先出)传输数据

叁歲伎倆 2022-12-24 15:49 248阅读 0赞

### 【python中级】 多进程之间通过Queue队列传输数据 ###

*  1.背景
 *  2.进程间传输数据
 *  3.进程间传输图片的bug
 *  4.解决进程间传输图片bug
 *  5. 总结

# 1.背景 #

应对爬虫：  
prompt: I only publish in csdn! 2020.11.29;  
**author**=“jn10010537”  
**describe**=“1754292866@qq.com”

[【python3】 list列表实现Queue队列先进先出的功能][python3_ list_Queue]  
https://blog.csdn.net/jn10010537/article/details/110289494  
**注意：**  
仅仅用列表实现了Queue队列的先进先出的功能，  
队列更多的功能并未实现，也不能在进程之间传输数据。

如下，list列表实现的队列不能在进程之间传输数据：

''' prompt:I only publish in csdn! 2020.11.29; '''
    __author__="jn10010537"
    __describe__="1754292866@qq.com"
    import sys
    import os
    from multiprocessing import Process
    
    class Queue(object):
        def __init__(self,size):
            self.MaxSize=size
            self.List=[]
    
        def __str__(self):
            return str(self.List)
    
        def Size(self):
            ''' 获取队列的当前长度 '''
            return len(self.List)
    
        def enqueue(self,items):
            ''' 入队，如果列队满了返回-1， 否则将元素插入队列尾 '''
            print("入队的pid:", os.getppid())
            if self.isfull():
                return -1
            self.List.append(items)
    
        def dequeue(self):
            ''' 出队，如果队列是空的返回-1， 否则返回队列头元素并将其从队列中移除该元素 '''
            print("出队的pid:", os.getppid())
            if self.isempty():
                return -1
            firstElement=self.List[0]
            self.List.remove(firstElement)
            return firstElement
    
        def isfull(self):
            ''' 判断队列是否满 '''
            return True if len(self.List)==self.MaxSize else False
    
        def isempty(self):
            ''' 判断队列为空 '''
            return True if len(self.List)==0 else False
    
    def fuc1(queue):
        print()
        print("************获得 fuc1 的pid:", os.getppid())
        print(">>>数据入队...")
        queue.enqueue("你好，jn10010537!")
        queue.enqueue("遇见jn10010537，很高兴!")
        print("队列里的元素：",queue)
        print(">>>数据入队完毕!\n")
    
        print(">>>数据出队:",queue.dequeue())
        print("队列里的元素：", queue)
        print(">>>数据出入队完毕!\n")
    
    def fuc2(queue):
        print()
        print("************获得 fuc2 的pid:", os.getppid())
        print("队列里的元素：", queue)
        print(">>>数据出队:", queue.dequeue())
        print(">>>数据出入队完毕!\n")
    
    
    if __name__=="__main__":
        print("python的版本:",sys.version)
        print("获得父进程的pid:",os.getppid())
        # 实例化自定义的队列
        queue =Queue(1000)
        #在主进程中开两个子进程
        run_fuc1 = Process(target=fuc1, args=(queue,))
        run_fuc1.start()
        run_fuc1.join()
        run_fuc2 = Process(target=fuc2, args=(queue,))
        run_fuc2.start()
        run_fuc2.join()
        # 打印进程的 id
        print("run_fuc1.pid:", run_fuc1.pid)
        print("run_fuc2.pid:", run_fuc2.pid)

运行结果如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2puMTAwMTA1Mzc_size_16_color_FFFFFF_t_70]  
**可以看到用List实现的队列class类是不能在两个子进程之间传输数据的。**

本博客  
通过python的multiprocessing.Queue()在两个子进程之间传输数据，并分析这个过程中传输的bug,以及解决办法。

# 2.进程间传输数据 #

使用python的multiprocessing.Queue()先进先出队列  
在两个子进程之间传输字符数据：

''' prompt:I only publish in csdn! 2020.11.29; '''
    __author__="jn10010537"
    __describe__="1754292866@qq.com"
    
    from multiprocessing import Process,Queue
    import os
    import sys
    def putData(queue):
        print("\n************获得 putData 的pid:", os.getppid())
        print(">>>数据入队...")
        queue.put("你好，jn10010537!")
        queue.put("遇见jn10010537，很高兴!")
        print(">>>数据入队完毕！")
    
    def getData(queue):
        print("\n************获得 getData 的pid:", os.getppid())
        print("队列里的元素：", queue.queue)
        while not queue.empty():
            print(">>>数据出队:", queue.get())
        print(">>>数据出入队完毕!\n")
    
    
    if __name__ == '__main__':
        print("python的版本:", sys.version)
        print("获得父进程的pid:", os.getppid())
    
        # 如果Queue(N)不指定长度,默认最大(和硬件相关)
        queue = Queue()
    
        run_putData = Process(target=putData,args=(queue,))
        run_putData.start()
        run_putData.join()
    
        run_getData = Process(target=getData, args=(queue,))
        run_getData.start()
        run_getData.join()
    
        # 打印进程的 id
        print("run_putData.pid:", run_putData.pid)
        print("run_getData.pid:", run_getData.pid)

运行如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2puMTAwMTA1Mzc_size_16_color_FFFFFF_t_70 1]

# 3.进程间传输图片的bug #

注意：图片也属于数据，这里单独拎出来，讲传输相对大的数据需要的问题：

''' prompt:I only publish in csdn! 2020.11.29; '''
    __author__="jn10010537"
    __describe__="1754292866@qq.com"
    
    from multiprocessing import Process,Queue
    import os
    import cv2
    import sys
    def putData(queue):
        print("\n************获得 putData 的pid:", os.getppid())
        print(">>>数据入队...")
        img = cv2.imread("1111.png")
        queue.put(img)
        print(">>>数据入队完毕！")
    
    def getData(queue):
        print("\n************获得 getData 的pid:", os.getppid())
        print("queue.empty()=", queue.empty())
        while not queue.empty():
            print(">>>数据出队中......")
            data = queue.get()
            cv2.imshow("picture", data)
            cv2.waitKey(0)
        print(">>>数据出入队完毕!\n")
    
    if __name__ == '__main__':
        print("python的版本:", sys.version)
        print("获得父进程的pid:", os.getppid())
    
        # 如果Queue(N)不指定长度,默认最大(和硬件相关)
        queue = Queue()
    
        run_putData = Process(target=putData,args=(queue,))
        run_putData.start()
        run_putData.join()
    
        run_getData = Process(target=getData, args=(queue,))
        run_getData.start()
        run_getData.join()
    
        # 打印进程的 id
        print("run_putData.pid:", run_putData.pid)
        print("run_getData.pid:", run_getData.pid)

运行结果：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2puMTAwMTA1Mzc_size_16_color_FFFFFF_t_70 2]  
如上图，在run\_getData进程中无法获得图片数据。

# 4.解决进程间传输图片bug #

**修改：**  
\-1.getData函数中增加延时；  
\-2.进程开始与阻塞改成：run\_putData，run\_getData子进程全部开启之后，再阻塞。

修改后代码如下：

''' prompt:I only publish in csdn! 2020.11.29; '''
    __author__="jn10010537"
    __describe__="1754292866@qq.com"
    
    from multiprocessing import Process,Queue
    import os
    import cv2
    import sys
    import time
    def putData(queue):
        print("\n************获得 putData 的pid:", os.getppid())
        print(">>>数据入队...")
        img = cv2.imread("1111.png")
        queue.put(img)
        print(">>>数据入队完毕！")
    
    def getData(queue):
        time.sleep(0.0001)
        print("\n************获得 getData 的pid:", os.getppid())
        print("queue.empty()=", queue.empty())
        while not queue.empty():
            print(">>>数据出队中......")
            data = queue.get()
            cv2.imshow("picture", data)
            cv2.waitKey(0)
        print(">>>数据出入队完毕!\n")
    
    if __name__ == '__main__':
        print("python的版本:", sys.version)
        print("获得父进程的pid:", os.getppid())
    
        # 如果Queue(N)不指定长度,默认最大(和硬件相关)
        queue = Queue()
    
        run_putData = Process(target=putData,args=(queue,))
        run_getData = Process(target=getData, args=(queue,))
    
    
        run_putData.start()
        run_getData.start()
        
        # 打印进程的 id
        print("run_putData.pid:", run_putData.pid)
        print("run_getData.pid:", run_getData.pid)
        run_putData.join()
        run_getData.join()

运行结果如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2puMTAwMTA1Mzc_size_16_color_FFFFFF_t_70 3]

# 5. 总结 #

\-5.1 两个子进程数据交互，阻塞应该在两个子进程开启之后，再阻塞。  
\-5.2 在子进程getData函数中增加延时，避免queue.empty()为空的阻塞；  
即 进队列，和出队列是分开的：  
如果数据大一些，可能出现数据已经进队列了，出队列部分，队列检测依然为空！可以通过在出队列的函数中增加延时来避免；

出现这种情况（进队列，和出队列的queue.empty()不一致），代码如下：

''' prompt:I only publish in csdn! 2020.11.29; '''
    __author__="jn10010537"
    __describe__="1754292866@qq.com"
    
    from multiprocessing import Process,Queue
    import os
    import cv2
    import sys
    import time
    def putData(queue):
        print("\n************获得 putData 的pid:", os.getppid())
        print(">>>数据入队...")
        img = cv2.imread("1111.png")
        queue.put(img)
        print(">>>数据入队完毕！")
    
    def getData(queue):
        # time.sleep(0.0001)
        print("\n************获得 getData 的pid:", os.getppid())
        print("queue.empty()=", queue.empty())
    
        data = queue.get()
        cv2.imshow("AAAA", data)
        cv2.waitKey(0)
    
        while not queue.empty():
            print(">>>数据出队中......")
            data = queue.get()
            cv2.imshow("picture", data)
            cv2.waitKey(0)
        print(">>>数据出入队完毕!\n")
    
    if __name__ == '__main__':
        print("python的版本:", sys.version)
        print("获得父进程的pid:", os.getppid())
    
        # 如果Queue(N)不指定长度,默认最大(和硬件相关)
        queue = Queue()
    
        run_putData = Process(target=putData,args=(queue,))
        run_getData = Process(target=getData, args=(queue,))
    
    
        run_putData.start()
        run_getData.start()
    
        # 打印进程的 id
        print("run_putData.pid:", run_putData.pid)
        print("run_getData.pid:", run_getData.pid)
        run_putData.join()
        run_getData.join()

运行如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2puMTAwMTA1Mzc_size_16_color_FFFFFF_t_70 4]

[python3_ list_Queue]: https://blog.csdn.net/jn10010537/article/details/110289494
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2puMTAwMTA1Mzc_size_16_color_FFFFFF_t_70]: /images/20221120/fe858481c5e746d7861c29f7b158e5d7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2puMTAwMTA1Mzc_size_16_color_FFFFFF_t_70 1]: /images/20221120/e9dd68b101c3497aa22b3e807881bc7f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2puMTAwMTA1Mzc_size_16_color_FFFFFF_t_70 2]: /images/20221120/3b120ab97dc04e0f90d6cecc07702326.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2puMTAwMTA1Mzc_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20201129143733271.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2puMTAwMTA1Mzc=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2puMTAwMTA1Mzc_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/20201129145618927.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2puMTAwMTA1Mzc=,size_16,color_FFFFFF,t_70