1

Вот мой код, который иллюстрирует проблему:

from threading import Thread
import queue
import time

q = queue.Queue()


class Worker(Thread):
    def __init__(self, name):
        Thread.__init__(self)
        self.name = name

    def run(self):
        if self.name == '0':
            update_tasks()
        else:
            while True:
                task = q.get()
                result = do_work(task)
                if not result:
                    q.put(task)
                
            
def update_tasks():
    while True:
        tasks = set(get_new_tasks()) - set(get_completed_tasks) - set(list(q.queue))
        for task in tasks:
            q.put(task)
        time.sleep(10)


def run_threads():
    threads = []
    for i in range(15):
        my_thread = Worker(str(i))
        my_thread.start()
        threads.append(my_thread)
    for thread in threads:
        thread.join()

run_threads()

Я запускаю 15 потоков. Из них 1 поток я отправляю пополнять количество задач функцией update_tasks() А остальные 14 должны решать эти задачи функцией do_work() Новые задачи поступают каждые 10 секунд. При этом, в свежеполученных задачах постоянно приходят те, что уже выполнены. Поэтому функция update_tasks() сравнивает свежеполученные задачи с теми, что уже сделаны, и теми, что находятся сейчас в очереди.

А теперь сама проблема: do_work() решает задачу примерно за 1 минуту и 14 задач, которые сейчас в работе снова поступают в список новых задач. Потому что они еще не появились в списке решенных задач. Но и в q.queue их уже нет. Эта конструкция их пропускает:

asks = set(get_new_tasks()) - set(get_completed_tasks) - set(list(q.queue))

Подскажите, пожалуйста, как можно одним потоком получить задачи, которые в это время выполняют другие потоки, для того чтобы избежать повторного выполнения одних и тех же задач?

3
  • 2
    Может попробуешь драматик или селери? Все таки готовые решения для организации очереди задач. А так, в твоем псевдо брокере должно быть 3 набора: задачи готовые к исполнению, задачи в процессе исполнения, выполненные задачи. Каждая задача отдается работнику и ожидается что работник вернет ответ в течение N времени, если он этого не сделал, то задача публикуется повторно. Каждый работник должен отчитываться брокеру о том что он закончил. Примерно...
    – alex
    22 ноя 2020 в 23:48
  • @alex А что быстрее в освоении. Драматик или Селери? Мне чем проще, тем лучше. 23 ноя 2020 в 0:50
  • Наверное драматик.
    – alex
    23 ноя 2020 в 10:37

1 ответ 1

1

исходя из вашего кода, зачем вообще проверять выполнение задач в воркерах, не проще ли делать это в update_tasks, она ведь и так знает что отправляла воркерам, а безрезультатные task добавляет сам воркер

def update_tasks():
    _tasks = set()  # ожидающие выполнения или уже выполненные task

    while True:
        tasks = get_new_tasks() - _tasks  # только новые task

        if tasks:
            _tasks.update(tasks)
            for task in tasks:
                q.put(task)
1
  • Благодарю за помощь. Я приблизительно так и сделал. Но так-же я понял, что лучше не изобретать велосипед на костылях, по этому начал знакомится с брокерами задач. 25 ноя 2020 в 10:55

Ваш ответ

By clicking “Отправить ответ”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.