python 进程间通信 Queue()、Pipe()、manager.list()、manager.dict()、manager.Queue()

👨‍💻个人简介： 深度学习图像领域工作者
🎉总结链接：
             链接中主要是个人工作的总结，每个链接都是一些常用demo，代码直接复制运行即可。包括：
                    📌1.工作中常用深度学习脚本
                    📌2.torch、numpy等常用函数详解
                    📌3.opencv 图片、视频等操作
                    📌4.个人工作中的项目总结（纯干活）
              链接： https://blog.csdn.net/qq_28949847/article/details/128552785
🎉视频讲解： 以上记录，通过B站等平台进行了视频讲解使用，可搜索 ‘Python图像识别’ 进行观看
              B站：Python图像识别
              抖音：Python图像识别
              西瓜视频：Python图像识别

---

1. multiprocessing.Queue()

示例代码：（主要是测试下通信时间）

结果： 通信传输的数据是一张 4K 的 图片， 数据比较大， put 基本上不消耗时间， get 平均耗时在 30ms

import timeimport numpy as np
import multiprocessingdef test(q):img = np.random.random([2160, 3840, 3])img = img.astype(np.uint8)t2 = time.time()# img = cv2.resize(img, (1920, 1080))for i in range(100):q.put(img)print('Time:', time.time() - t2)if __name__ == '__main__':q = multiprocessing.Queue()  # 定义queue,这里可以传入参数，即队列所含最大元素量p1 = multiprocessing.Process(target=test, args=(q,))  # 多进程p1.start()while True:  # 判断队列是否为非空，非空才继续执行下一步if q.qsize() != 0:t1 = time.time()res = q.get()  # 获取一个元素print('res:', time.time() - t1)

2. multiprocessing.Pipe

示例代码： put 和 get 基本上都是在30ms

import multiprocessing
import timeimport numpy as npdef test1(conn):img = np.random.random([2160, 3840, 3])img = img.astype(np.uint8)for i in range(100):t1 = time.time()conn.send(img)print('send:', time.time() - t1)if __name__ == '__main__':p_conn, c_conn = multiprocessing.Pipe()  # 定义一个管道的两端，之后将这两端传出去p1 = multiprocessing.Process(target=test1, args=(p_conn,))  # 把一端给test1p1.start()while True:time.sleep(5)for i in range(100):t1 = time.time()s = c_conn.recv()print('recv:', time.time() - t1)

3. manager.list()、manager.dict()、manager.Queue()

import multiprocessing
import timeimport numpy as npdef test_dic(dict1):img = np.random.random([2160, 3840, 3])img = img.astype(np.uint8)t1 = time.time()dict1['name1'] = imgprint(time.time() - t1)def test_lst(list1):img = np.random.random([2160, 3840, 3])img = img.astype(np.uint8)t1 = time.time()list1.append(img)print('ssss:', time.time() - t1)def test_queue1(queue1):img = np.random.random([2160, 3840, 3])img = img.astype(np.uint8)t1 = time.time()queue1.put(img)print('ssss:', time.time() - t1)if __name__ == '__main__':with multiprocessing.Manager() as manager:  # 引入managerlist1 = manager.list()  # 设置list1和dict1为共享对象dict1 = manager.dict()queue1 = manager.Queue()# p = multiprocessing.Process(target=test_dic, args=(dict1,))p = multiprocessing.Process(target=test_lst, args=(list1,))# p = multiprocessing.Process(target=test_queue1, args=(queue1,))p.start()t1 = time.time()# 0.030# s = dict1['name1']# 0.030ss = list1[0]# 0.033s# s = queue1.get()print('llll:', time.time() - t1)