如何终止带有队列的Python进程+ 队列大小的错误

该问题在"加入使用队列的流程"部分的https://docs.python.org/3/library/multiprocessing.html#multiprocessing-programming中进行了描述.

基本思想是，流程将等待然后终止，直到所有缓冲的项目由"feed"线程提供到底层管道.

import multiprocessing
import random
import time


def process1(data, queue1):
    for item in data:
        queue1.put(item)

    queue1.put(None)


def process2(queue1, queue2):
    while True:
        item = queue1.get()
        if item is None:
            queue2.put(None)
            break
        queue2.put(item * 2)


def process3(queue2, queue3):
    while True:
        item = queue2.get()
        if item is None:
            queue3.put(None)
            break
        queue3.put(item * 2)


def process4(queue3, queue4):
    # calling queue4.cancel_join_thread() will correctly terminate the process
    # queue4.cancel_join_thread()
    # BUT not all items are put into the queue4 (rest are discarded)

    # see https://docs.python.org/3/library/multiprocessing.html#multiprocessing-programming
    # section "Joining processes that use queues"

    while True:
        item = queue3.get()
        if item is None:
            queue4.put(None)
            print("None received")
            break
        queue4.put(item * 2)
    print("End process4")
    print(queue4.qsize())


if __name__ == "__main__":
    start_time = time.time()

    data = [random.randint(1, 10000) for _ in range(10000)]

    queue1 = multiprocessing.Queue()
    queue2 = multiprocessing.Queue()
    queue3 = multiprocessing.Queue()
    queue4 = multiprocessing.Queue()

    p1 = multiprocessing.Process(target=process1, args=(data, queue1))
    p2 = multiprocessing.Process(target=process2, args=(queue1, queue2))
    p3 = multiprocessing.Process(target=process3, args=(queue2, queue3))
    p4 = multiprocessing.Process(target=process4, args=(queue3, queue4))

    p1.start()
    p2.start()
    p3.start()
    p4.start()
    print("Proc started")

    p1.join()
    print("p1 join")
    p2.join()
    print("p2 join")
    p3.join()
    print("p3 join")

    # don't call p4.join() here
    # p4.join()
    # print("p4 join")

    results = []
    print("While loop")
    while True:
        item = queue4.get()
        if item is None:
            break
        results.append(item)

    end_time = time.time()

    # call p4.join() here instead:
    p4.join()
    print("p4 join")

    print(end_time - start_time)

输出:

Proc started
p1 join
None received
End process4
10001
p2 join
p3 join
While loop
p4 join
0.1303730010986328