Вот мой код, который иллюстрирует проблему:
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))
Подскажите, пожалуйста, как можно одним потоком получить задачи, которые в это время выполняют другие потоки, для того чтобы избежать повторного выполнения одних и тех же задач?