Как работает Queue в Python

Question

тестовый код очень прост:

import threading, Queue
import time, random

class Worker(threading.Thread):
    def __init__(self, index, queue):
        threading.Thread.__init__(self)
        self.index = index
        self.queue = queue
    def run(self):
        while 1:
            time.sleep(random.random())
            item = self.queue.get()
            if item is None:
                break
            print "index:", self.index, "task", item, "finished"
            self.queue.task_done()

queue = Queue.Queue(0)
for i in range(2):
    Worker(i, queue).start()
for i in range(10):
    queue.put(i)
for i in range(2):
    queue.put(None)
print "Main OK"

и результат немного меняется каждый раз, когда я запускаю его, вот только один:

Main OK
index: 1 task 0 finished
index: 0 task 1 finished
index: 0 task 2 finished
index: 1 task 3 finished
index: 1 task 4 finished
index: 0 task 5 finished
index: 1 task 6 finished
index: 0 task 7 finished
index: 1 task 8 finished
index: 1 task 9 finished

IMO, когда основной поток завершается, будет напечатано «Main OK», затем будет выполняться первый поток, пока он не войдет в time.sleep(random.random()), затем первый поток перейдет в спящий режим, а второй поток продолжится. То же самое относится к первому потоку, второй поток будет находиться в спящем режиме при входе в time.sleep(random.random()), затем первый поток продолжится снова. и он напечатает index:0 task 0 finished сразу после Main OK, но в действительности то, что следует за Main OK, равно index: 1..., а не index: 0...! Зачем? и кажется, что очередь не работает как обычный многопоточный, иногда один и тот же индексный поток будет выполняться три раза или более непрерывно! В каком мире работает механизм очереди? любая помощь приветствуется!

Answer 1

Нет гарантий, в каком порядке будут выполняться потоки.Если бы была гарантия заказа, это потребовало бы большой синхронизации и, следовательно, было бы эквивалентно сериализованной производительности.

Также вы должны использовать queue.join() для ожидания (блокировки) всех рабочих до конца.

Answer 2

У вас есть три темы;две рабочие темы и одна основная.Все три работают одновременно, практически.То, что вы предлагаете (это четкое переключение в точках, которые вы будете знать заранее), не соответствует действительности.Вы запускаете рабочие потоки до того, как очередь заполняется, поэтому они сразу же запускаются в режим сна ().Затем вы заполняете очередь.Весьма вероятно, что один из потоков выйдет из режима сна () раньше, чем другой, и получит первый элемент из очереди, обработает его (напечатает) и снова перейдет в следующий режим сна ().Вполне возможно (из-за случайного выбора), что первый рабочий будет спать 0,01 с каждый раз, в то время как другой спит в течение 0,4 с с начала, тогда все элементы будут обработаны первым процессом.

Если больше, чемодин рабочий поток блокируется в методе Queue.get () (может произойти, только если оба оставили свой сон (), пока очередь еще не заполнена), у вас нет детерминизма, по которому рабочий поток пробуждается для обработки элемента.

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

def run(self):
    if self.index == 0:
        time.sleep(0.1)
    while 1:
        time.sleep(0.2)
        item = self.queue.get()
        if item is None:
            break
        print "index:", self.index, "task", item, "finished"
        self.queue.task_done()

Answer 3

Хотя порядок выполнения потоков не гарантирован, у вас также есть time.sleep(random.random()).

>>> random.random()
0.044693605707810558
>>> random.random()
0.16270424255105465
>>> random.random()
0.74068552817650446