Моделирование токенов отмены в Python Threading

Кажется, вы создаете новый поток для каждой задачи из очереди. Это само по себе расточительно, а также приводит к проблеме ограничения количества потоков.

Вместо этого распространенный подход заключается в создании фиксированного числа рабочих потоков и предоставлении им возможности свободно извлекать задачи из очереди. Чтобы отменить очередь, вы можете очистить ее и оставить рабочие процессы в ожидании будущей работы.

Я думаю, что ваш код можно упростить, используя отравление и Thread.join():

from Queue import Queue
from threading import Thread

poison = object()

class TaskQueue(object):

    def __init__(self, limit):
        def process_items():
            while True:
                callback = self._queue.get()
                if callback is poison:
                    break
                try:
                    callback()
                except:
                    pass
                finally:
                    self._queue.task_done()
        self._workers = [Thread(target=process_items) for _ in range(limit)]
        self._queue = Queue()

    def enqueue(self, callback):
        self._queue.put(callback)

    def start(self):
        for worker in self._workers:
            worker.start()

    def stop(self):
        for worker in self._workers:
            self._queue.put(poison)
        while self._workers:
            self._workers.pop().join()

Непроверенный.

Я удалил комментарии, для краткости.

Кроме того, в этой версии process_items() действительно приватный.

Кстати: Весь смысл модуля Queue в том, чтобы освободить вас от ужасных блокировок и событий.

Я последовал совету Janne Karila и создал пул потоков. Это устранило необходимость в семафоре. Проблема в том, что если вы когда-нибудь ожидаете, что очередь исчезнет, ​​вам придется остановить выполнение рабочих потоков (просто вариант того, что я делал раньше). Новый код довольно похож:

class TaskQueue(object):
    """
    Queues tasks to be run in separate threads and limits the number
    concurrently running tasks.

    """

    def __init__(self, limit):
        """Initializes a new instance of a TaskQueue."""
        self.__workers = []
        for _ in range(limit):
            worker = Thread(target=self.__process_items)
            self.__workers.append(worker)
        self.__queue = Queue()
        self.__cancelled = False
        self.__lock = Lock()
        self.__event = Event()

    def enqueue(self, callback):
        """Indicates that the given callback should be ran."""
        self.__queue.put(callback)

    def start(self):
        """Tells the task queue to start running the queued tasks."""
        for worker in self.__workers:
            worker.start()

    def stop(self):
        """
        Stops the queue from processing anymore tasks. Any actively running
        tasks will run to completion.

        """
        self.__cancel()
        # prevent blocking on a Queue.get
        for _ in range(len(self.__workers)):
            self.__queue.put(lambda: None)
            self.__event.wait()

    def __cancel(self):
        with self.__lock:
            self.__queue.queue.clear()
            self.__cancelled = True

    def __process_items(self):
        while True:
            callback = self.__queue.get()

            # see if the queue has been stopped before running the task
            if self.__is_canceled():
                break

            try:
                callback()
            except:
                pass
            finally:
                self.__queue.task_done()
        self.__event.set()

    def __is_canceled(self):
        with self.__lock:
            return self.__cancelled

Если вы посмотрите внимательно, мне пришлось провести некоторые расчеты, чтобы убить рабочих. В основном я жду Event столько раз, сколько рабочих. Я clear устанавливаю базовую очередь, чтобы предотвратить отмену воркеров любым другим способом. Я также жду после закачки каждого фиктивного значения в очередь, поэтому только один рабочий может отменить за раз.

Я провел несколько тестов на этом, и, похоже, он работает. Было бы неплохо устранить необходимость в фиктивных значениях.