Помогите разобраться с python threading thread queue, срочно надо!

[Арагорн]

Всем привет! Нужна помощь с многопоточной программой на Python. Пытаюсь использовать threading и queue, но что-то идет не так. Создаю потоки, кидаю задачи в очередь, но они работают не так, как я ожидал. Кому не лень, гляньте на мой пример кода и подскажите, что я упускаю:

      import threading

      import queue

      

      def worker(q):

          while True:

              task = q.get()

              if task is None:

                  break

              print("Работаем над:", task)

              q.task_done()

      

      task_queue = queue.Queue()

      

      threads = []

      for i in range(4):

          t = threading.Thread(target=worker, args=(task_queue,))

          t.start()

          threads.append(t)

      

      for task in range(10):

          task_queue.put(task)

      

      task_queue.join()

      

      for i in range(4):

          task_queue.put(None)

      for t in threads:

          t.join() 

Друзья, в чем может быть дело? Почему мой код странно себя ведет?

[admin]

Ждём вас в нашем чате в Телеграмм ==>> @pythoneer_chat

А ТАКЖЕ: Канал о Python, статьи и книги ==>> @pythoneer_ru

[Vadim123]

Здорово! На первый взгляд, твой код выглядит норм, но можно немного улучшить. Например, ты правильно использовал join() для ожидания завершения задач, но проблема может быть в неправильной синхронизации потоков. Попробуй использовать условные переменные или Lock объекты для лучшей синхронизации.

[GalaxySage]

Здорово! На первый взгляд, твой код выглядит норм, но можно немного улучшить. Например, ты правильно использовал join() для ожидания завершения задач, но проблема может быть в неправильной синхронизации потоков. Попробуй использовать условные переменные или Lock объекты для лучшей синхронизации.

Крутая идея с Lock! Надо будет попробовать, вдруг решит проблему.

[DragonSpirit007]

Йо! Тут все вроде нормалек, но тебе лучше явно обрабатывать случаи, когда очередь пуста. Используй q.empty() или добавь таймаут при получении задачи q.get(timeout=1). Иногда потоки просто ждут, если очередь пуста. Надеюсь, это испавит трабл.

[TaleWarrior909]

Йо! Тут все вроде нормалек, но тебе лучше явно обрабатывать случаи, когда очередь пуста. Используй q.empty() или добавь таймаут при получении задачи q.get(timeout=1). Иногда потоки просто ждут, если очередь пуста. Надеюсь, это испавит трабл.

Да, про таймаут я не подумал, тогда бы потоки не зависали один-на-один. Спасибо за идею!

[Galactus]

Эй, коллега! Мне кажется, твои потоки лагают из-за того, что слишком много задач поступает одновременно. Подумай использовать ThreadPoolExecutor из concurrent.futures, он проще в использовании и может уменьшить твои боли. Интерфейс у него гораздо удобнее и короче.

[Егор]

Эй, коллега! Мне кажется, твои потоки лагают из-за того, что слишком много задач поступает одновременно. Подумай использовать ThreadPoolExecutor из concurrent.futures, он проще в использовании и может уменьшить твои боли. Интерфейс у него гораздо удобнее и короче.

Ага, возможно, ThreadPoolExecutor как раз мне и нужен. Спасибо за наводку!

[Петр Сергеевич]

Привет! Посмотрел твой код и у меня мысли такие: попробуй добавить более детальный logging, чтобы понять в какие моменты возникают проблемы. Это поможет отловить редкие баги или узнать, есть ли вообще проблема с потоками. Иногда простое логирование творит чудеса.

import threading

import queue

import logging



logging.basicConfig(level=logging.DEBUG, format="%(threadName)s: %(message)s")



def worker(q):

    while True:

        task = q.get()

        if task is None:

            break

        logging.debug(f'Работаем над: {task}')

        q.task_done()



task_queue = queue.Queue()



threads = []

for i in range(4):

    t = threading.Thread(target=worker, args=(task_queue,))

    t.start()

    threads.append(t)



for task in range(10):

    task_queue.put(task)



task_queue.join()



for i in range(4):

    task_queue.put(None)

for t in threads:

    t.join() 

Может это прояснит пару моментов. Чекни и отпишись, что получилось.

[NinaSun]

Привет! Посмотрел твой код и у меня мысли такие: попробуй добавить более детальный logging, чтобы понять в какие моменты возникают проблемы. Это поможет отловить редкие баги или узнать, есть ли вообще проблема с потоками. Иногда простое логирование творит чудеса.

import threading

import queue

import logging



logging.basicConfig(level=logging.DEBUG, format="%(threadName)s: %(message)s")



def worker(q):

    while True:

        task = q.get()

        if task is None:

            break

        logging.debug(f'Работаем над: {task}')

        q.task_done()



task_queue = queue.Queue()



threads = []

for i in range(4):

    t = threading.Thread(target=worker, args=(task_queue,))

    t.start()

    threads.append(t)



for task in range(10):

    task_queue.put(task)



task_queue.join()



for i in range(4):

    task_queue.put(None)

for t in threads:

    t.join() 

Может это прояснит пару моментов. Чекни и отпишись, что получилось.

Йо, точно! Логирование может показать, где факап! Сто пудов что-то всплывет, что мы просто не замечаем!