Почему производитель-потребитель не останавливается?

Question

Я нашел пример, представляющий производителя-потребителя с двумя потоками. Но когда я посылаю сигнал процессу, чтобы остановить, это не так. Ожидается второй сигнал, например SIGKILL чтобы полностью остановиться. Я думал, что проблема с task_done(), но, похоже, нет.

import time

import queue
import threading
import random


class Producer(threading.Thread):
    """
    Produces random integers to a list
    """

    def __init__(self, queue):
        """
        Constructor.

        @param queue queue synchronization object
        """
        threading.Thread.__init__(self)
        self.queue = queue

    def run(self):
        """
        Thread run method. Append random integers to the integers
        list at random time.
        """
        while True:
            integer = random.randint(0, 256)
            self.queue.put(integer)
            print('%d put to queue by %s' % (integer, self.name))
            time.sleep(1)


class Consumer(threading.Thread):
    """
    Consumes random integers from a list
    """

    def __init__(self, queue):
        """
        Constructor.

        @param integers list of integers
        @param queue queue synchronization object
        """
        threading.Thread.__init__(self)
        self.queue = queue

    def run(self):
        """
        Thread run method. Consumes integers from list
        """
        while True:
            integer = self.queue.get()
            print('%d popped from list by %s' % (integer, self.name))
            self.queue.task_done()


def main():
    q = queue.Queue()
    t1 = Producer(q)
    t2 = Consumer(q)
    t1.start()
    t2.start()
    t1.join()
    t2.join()


if __name__ == '__main__':
    main()

Выход:

210 put to queue by Thread-1
210 popped from list by Thread-2
Traceback (most recent call last):
  File "/Users/abc/PycharmProjects/untitled1/ssid.py", line 74, in <module>
    main()
  File "/Users/abc/PycharmProjects/untitled1/ssid.py", line 69, in main
    t1.join()
  File "/usr/local/Cellar/python3/3.6.3/Frameworks/Python.framework/Versions/3.6/lib/python3.6/threading.py", line 1056, in join
    self._wait_for_tstate_lock()
  File "/usr/local/Cellar/python3/3.6.3/Frameworks/Python.framework/Versions/3.6/lib/python3.6/threading.py", line 1072, in _wait_for_tstate_lock
    elif lock.acquire(block, timeout):
KeyboardInterrupt
244 put to queue by Thread-1
244 popped from list by Thread-2
85 put to queue by Thread-1
85 popped from list by Thread-2
160 put to queue by Thread-1
160 popped from list by Thread-2

Answer 1

Это потому, что только получение основного потока остановлено KeyboardInterrupt. Вы можете наблюдать это, позволяя вашим дочерним потокам печатать threading.enumerate(), который возвращает все живые потоки + основной поток.

import time
import queue
import threading
import random


class Producer(threading.Thread):

    def __init__(self, queue):
        super().__init__()
        self.queue = queue

    def run(self):
        while True:
            integer = random.randint(0, 256)
            self.queue.put(integer)
            print(f'{integer} put to queue by {self.name} '
                  f'threads: {threading.enumerate()}')
            time.sleep(1)


class Consumer(threading.Thread):

    def __init__(self, queue):
        super().__init__()
        self.queue = queue

    def run(self):
        while True:
            integer = self.queue.get()
            print(f'{integer} popped from list by {self.name} '
                  f'threads:{threading.enumerate()}')
            self.queue.task_done()


def main():
    q = queue.Queue()
    t1 = Producer(q)
    t2 = Consumer(q)
    # t1.daemon = True
    # t2.daemon = True
    t1.start()
    t2.start()
    t1.join()
    t2.join()


if __name__ == '__main__':
    try:
        main()
    except KeyboardInterrupt:
        print('got KeyboardInterrupt')

Пример вывода с KeyboardInterrupt. Обратите внимание, что MainThread указан как «остановленный» после KeyboardInterrupt:

97 put to queue by Thread-1 threads: [<_MainThread(MainThread, started 
139810293606208)>, <Producer(Thread-1, started 139810250913536)>, 
<Consumer(Thread-2, started 139810242520832)>]
97 popped from list by Thread-2 threads:[<_MainThread(MainThread, started 
139810293606208)>, <Producer(Thread-1, started 139810250913536)>, 
<Consumer(Thread-2, started 139810242520832)>]
got KeyboardInterrupt
92 put to queue by Thread-1 threads: [<_MainThread(MainThread, stopped 
139810293606208)>, <Producer(Thread-1, started 139810250913536)>, 
<Consumer(Thread-2, started 139810242520832)>]
92 popped from list by Thread-2 threads:[<_MainThread(MainThread, stopped 
139810293606208)>, <Producer(Thread-1, started 139810250913536)>, 
<Consumer(Thread-2, started 139810242520832)>]

Вы можете создать демонов дочерних потоков, чтобы они могли выходить с основным потоком. Но это следует учитывать только в том случае, если ваши потоки не содержат никаких ресурсов:

Примечание. Потоки демона внезапно останавливаются при завершении работы. Их ресурсы (такие как открытые файлы, транзакции базы данных и т. Д.) Могут быть освобождены неправильно. Если вы хотите, чтобы ваши потоки корректно останавливались, сделайте их недемоническими и используйте подходящий механизм сигнализации, такой как Event docs .

Лучшим способом было бы перехватить KeyboardInterrupt, как в приведенном выше коде, и отправить дозорное значение по очереди дочерним потокам, чтобы сообщить им, что они должны завершиться, что позволит им выполнить очистку перед выходом.