[已解决]multiprocessing.Process的提问

歌者文明清理员

1. # 获取进程的ID
   
2. from os import getpid
   
3. # 进程和队列
   
4. from multiprocessing import Process, Queue
   
5. # 本例子需要使用随机数
   
6. from random import randint
   
7. 
   
8. 
   
9. class Point:
   
10.     # 积分类
    
11.     def __init__(self, point):
    
12.         self.num = point
    
13. 
    
14.     def __repr__(self):
    
15.         return f'Point({self.num})'
    
16. 
    
17. 
    
18. def producer(queue):
    
19.     # 生产者函数
    
20.     print(f'生产者pid：{getpid()}')
    
21.     for i in range(10):
    
22.         point_num = randint(1, 10)
    
23.         point_part = Point(point_num)
    
24.         queue.put(point_part)
    
25. 
    
26. 
    
27. def consumer(queue):
    
28.     # 消费者
    
29.     print(f'消费者pid：{getpid()}')
    
30.     while True:
    
31.         point = queue.get()
    
32.         print(point)
    
33. 
    
34. 
    
35. que = Queue()
    
36. cons = Process(target=consumer, args=(que, ))
    
37. prod = Process(target=producer, args=(que, ))
    
38. cons.start()
    
39. prod.start()
    
40. cons.join()
    
41. prod.terminate()
    

复制代码

报错信息如下：

1. Traceback (most recent call last):
   
2.   File "<string>", line 1, in <module>
   
3. Traceback (most recent call last):
   
4.   File "<string>", line 1, in <module>
   
5.   File "c:\users\bruce\appdata\local\programs\python\python37\lib\multiprocessing\spawn.py", line 105, in spawn_main
   
6.   File "c:\users\bruce\appdata\local\programs\python\python37\lib\multiprocessing\spawn.py", line 105, in spawn_main
   
7.     exitcode = _main(fd)
   
8.     exitcode = _main(fd)
   
9.   File "c:\users\bruce\appdata\local\programs\python\python37\lib\multiprocessing\spawn.py", line 115, in _main
   
10.   File "c:\users\bruce\appdata\local\programs\python\python37\lib\multiprocessing\spawn.py", line 115, in _main
    
11.     self = reduction.pickle.load(from_parent)
    
12.     self = reduction.pickle.load(from_parent)
    
13. AttributeError: Can't get attribute 'consumer' on <module '__main__' (built-in)>
    
14. AttributeError: Can't get attribute 'producer' on <module '__main__' (built-in)>

复制代码

可能是因为需要if __name__ == '__main__'，所以我又更改了代码

1. # 获取进程的ID
   
2. from os import getpid
   
3. # 进程和队列
   
4. from multiprocessing import Process, Queue
   
5. # 本例子需要使用随机数
   
6. from random import randint
   
7. 
   
8. 
   
9. class Point:
   
10.     # 积分类
    
11.     def __init__(self, point):
    
12.         self.num = point
    
13. 
    
14.     def __repr__(self):
    
15.         return f'Point({self.num})'
    
16. 
    
17. 
    
18. def producer(queue):
    
19.     # 生产者函数
    
20.     print(f'生产者pid：{getpid()}')
    
21.     for i in range(10):
    
22.         point_num = randint(1, 10)
    
23.         point_part = Point(point_num)
    
24.         queue.put(point_part)
    
25. 
    
26. 
    
27. def consumer(queue):
    
28.     # 消费者
    
29.     print(f'消费者pid：{getpid()}')
    
30.     while True:
    
31.         point = queue.get()
    
32.         print(point)
    
33. 
    
34. 
    
35. if __name__ == '__main__':
    
36.     que = Queue()
    
37.     cons = Process(target=consumer, args=(que, ))
    
38.     prod = Process(target=producer, args=(que, ))
    
39.     cons.start()
    
40.     prod.start()
    
41.     cons.join()
    
42.     prod.terminate()
    

复制代码

虽然没有报错，也可以看到正常输出，但是 prod.terminate() 没用，死循环停不下来，我只能加上 timeout 参数并捕获错误：

1. def consumer(queue):
   
2.     # 消费者
   
3.     print(f'消费者pid：{getpid()}')
   
4.     while True:
   
5.         try:
   
6.             point = queue.get(timeout=1)
   
7.         except:
   
8.             break
   

复制代码

虽然问题已经解决，但为什么报错

最佳答案

月排行榜 / 总排行榜

isdkz

2023-4-8 14:44:41

根据您提供的信息，我理解了您遇到的两个问题。首先，让我们分别解决它们：

1、AttributeError: Can't get attribute 'consumer' on <module 'main' (built-in)>
这个错误是因为在使用multiprocessing模块时，Process实例需要在if __name__ == '__main__':块中创建。

这是因为multiprocessing模块使用子进程来并行执行任务，而子进程通常需要单独的独立的执行环境。

当您将Process实例移动到if __name__ == '__main__':代码块中时，这个问题就解决了。

2、死循环停不下来的问题
prod.terminate()没有停止死循环，原因是producer进程已经完成了任务并退出，而consumer进程在queue.get()时，由于队列为空，它会一直等待新的元素。

由于producer进程已经退出，队列不会再有新元素加入。因此，consumer进程会陷入死循环。

为了解决这个问题，您可以在producer函数完成任务后，向队列中放入一个特殊的结束标记。然后在consumer函数中检查队列中是否有这个结束标记，如果有，则跳出循环。这是一个示例：

1. def producer(queue):
   
2.     # 生产者函数
   
3.     print(f'生产者pid：{getpid()}')
   
4.     for i in range(10):
   
5.         point_num = randint(1, 10)
   
6.         point_part = Point(point_num)
   
7.         queue.put(point_part)
   
8.     queue.put(None)  # 放入结束标记
   
9. 
   
10. def consumer(queue):
    
11.     # 消费者
    
12.     print(f'消费者pid：{getpid()}')
    
13.     while True:
    
14.         point = queue.get()
    
15.         if point is None:  # 检查结束标记
    
16.             break
    
17.         print(point)

复制代码

这样，consumer函数就可以在检测到结束标记后正常退出循环了。

这样一来，您不再需要为queue.get()设置timeout参数，因为consumer进程会在处理完所有数据后自动退出。


完整代码：

1. # 获取进程的ID
   
2. from os import getpid
   
3. # 进程和队列
   
4. from multiprocessing import Process, Queue
   
5. # 本例子需要使用随机数
   
6. from random import randint
   
7. 
   
8. 
   
9. class Point:
   
10.     # 积分类
    
11.     def __init__(self, point):
    
12.         self.num = point
    
13. 
    
14.     def __repr__(self):
    
15.         return f'Point({self.num})'
    
16. 
    
17. 
    
18. def producer(queue):
    
19.     # 生产者函数
    
20.     print(f'生产者pid：{getpid()}')
    
21.     for i in range(10):
    
22.         point_num = randint(1, 10)
    
23.         point_part = Point(point_num)
    
24.         queue.put(point_part)
    
25.     queue.put(None)  # 放入结束标记
    
26. 
    
27. def consumer(queue):
    
28.     # 消费者
    
29.     print(f'消费者pid：{getpid()}')
    
30.     while True:
    
31.         point = queue.get()
    
32.         if point is None:  # 检查结束标记
    
33.             break
    
34.         print(point)
    
35. 
    
36. if __name__ == '__main__':
    
37.     que = Queue()
    
38.     cons = Process(target=consumer, args=(que, ))
    
39.     prod = Process(target=producer, args=(que, ))
    
40.     cons.start()
    
41.     prod.start()
    
42.     cons.join()
    
43.     prod.terminate()
    

复制代码

跳转到最佳答案楼层

isdkz

根据您提供的信息，我理解了您遇到的两个问题。首先，让我们分别解决它们：

1、AttributeError: Can't get attribute 'consumer' on <module 'main' (built-in)>
这个错误是因为在使用multiprocessing模块时，Process实例需要在if __name__ == '__main__':块中创建。

这是因为multiprocessing模块使用子进程来并行执行任务，而子进程通常需要单独的独立的执行环境。

当您将Process实例移动到if __name__ == '__main__':代码块中时，这个问题就解决了。

2、死循环停不下来的问题
prod.terminate()没有停止死循环，原因是producer进程已经完成了任务并退出，而consumer进程在queue.get()时，由于队列为空，它会一直等待新的元素。

由于producer进程已经退出，队列不会再有新元素加入。因此，consumer进程会陷入死循环。

为了解决这个问题，您可以在producer函数完成任务后，向队列中放入一个特殊的结束标记。然后在consumer函数中检查队列中是否有这个结束标记，如果有，则跳出循环。这是一个示例：

1. def producer(queue):
   
2.     # 生产者函数
   
3.     print(f'生产者pid：{getpid()}')
   
4.     for i in range(10):
   
5.         point_num = randint(1, 10)
   
6.         point_part = Point(point_num)
   
7.         queue.put(point_part)
   
8.     queue.put(None)  # 放入结束标记
   
9. 
   
10. def consumer(queue):
    
11.     # 消费者
    
12.     print(f'消费者pid：{getpid()}')
    
13.     while True:
    
14.         point = queue.get()
    
15.         if point is None:  # 检查结束标记
    
16.             break
    
17.         print(point)

复制代码

这样，consumer函数就可以在检测到结束标记后正常退出循环了。

这样一来，您不再需要为queue.get()设置timeout参数，因为consumer进程会在处理完所有数据后自动退出。


完整代码：

1. # 获取进程的ID
   
2. from os import getpid
   
3. # 进程和队列
   
4. from multiprocessing import Process, Queue
   
5. # 本例子需要使用随机数
   
6. from random import randint
   
7. 
   
8. 
   
9. class Point:
   
10.     # 积分类
    
11.     def __init__(self, point):
    
12.         self.num = point
    
13. 
    
14.     def __repr__(self):
    
15.         return f'Point({self.num})'
    
16. 
    
17. 
    
18. def producer(queue):
    
19.     # 生产者函数
    
20.     print(f'生产者pid：{getpid()}')
    
21.     for i in range(10):
    
22.         point_num = randint(1, 10)
    
23.         point_part = Point(point_num)
    
24.         queue.put(point_part)
    
25.     queue.put(None)  # 放入结束标记
    
26. 
    
27. def consumer(queue):
    
28.     # 消费者
    
29.     print(f'消费者pid：{getpid()}')
    
30.     while True:
    
31.         point = queue.get()
    
32.         if point is None:  # 检查结束标记
    
33.             break
    
34.         print(point)
    
35. 
    
36. if __name__ == '__main__':
    
37.     que = Queue()
    
38.     cons = Process(target=consumer, args=(que, ))
    
39.     prod = Process(target=producer, args=(que, ))
    
40.     cons.start()
    
41.     prod.start()
    
42.     cons.join()
    
43.     prod.terminate()
    

复制代码