Несколько очередей с пулами потоков не выполняют все задачи - PullRequest
Новая
       60

Несколько очередей с пулами потоков не выполняют все задачи

0 голосов
/ 17 апреля 2020

Цель состоит в том, чтобы иметь две очереди: первая очередь питает вторую очередь, и каждая очередь имеет различное ограничение скорости. Например, первый работник, работающий в первой очереди, должен выполнять 5 запросов каждые 5 секунд, а второй работник должен выполнять 2 запроса каждые две секунды. Или может быть ограничение скорости с 10 запросами каждые 2 секунды или 20 запросами каждые 3 секунды.

Мой код выполняется, однако он не завершает sh время, кажется, выключено. 

for x in range(20):
    first_queue.put(x)

def first_worker_job(queue):
    v = first_queue.get()
    log.info(f"first: {v}")
    second_queue.put(v)

def first_worker(first_queue):
    while True:
        with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
            furtureIteams = {
                executor.submit(first_worker_job, first_queue): index
                for index in range(5)
            }
            for future in concurrent.futures.as_completed(furtureIteams):
                f = furtureIteams[future]

            time.sleep(5)

def second_worker_job(queue):
    v = second_queue.get()
    log.info(f"second: {v}")

def second_worker(second_queue):

    while True:
        with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
            furtureIteams = {
                executor.submit(second_worker_job, first_queue): index
                for index in range(2)
            }
            for future in concurrent.futures.as_completed(furtureIteams):
                f = furtureIteams[future]

            time.sleep(2)

first_thread = threading.Thread(target=first_worker, args=(first_queue,))
first_thread.setDaemon(True)
first_thread.start()

second_thread = threading.Thread(target=second_worker, args=(second_queue,))
second_thread.setDaemon(True)
second_thread.start()

first_queue.join()
second_queue.join()

second_thread.join()
first_thread.join()

Вывод: 

[11:39:13] first: 1
[11:39:13] first: 2
[11:39:13] first: 3
[11:39:13] first: 4
[11:39:13] second: 0
[11:39:13] second: 1

[11:39:15] second: 2
[11:39:15] second: 3

[11:39:17] second: 4

[11:39:18] first: 5
[11:39:18] first: 6
[11:39:18] first: 7
[11:39:18] first: 8
[11:39:18] first: 9
[11:39:18] second: 5

[11:39:20] second: 6
[11:39:20] second: 7

[11:39:22] second: 8
[11:39:22] second: 9

[11:39:23] first: 10
[11:39:23] first: 11
[11:39:23] first: 12
[11:39:23] first: 13
[11:39:23] first: 14

[11:39:24] second: 10
[11:39:24] second: 11

[11:39:26] second: 12
[11:39:26] second: 13

[11:39:28] first: 15
[11:39:28] first: 16
[11:39:28] first: 17
[11:39:28] second: 14
[11:39:28] second: 15
[11:39:28] first: 18
[11:39:28] first: 19

[11:39:30] second: 16
[11:39:30] second: 17

[11:39:32] second: 18
[11:39:32] second: 19

Это то, что я ожидал: 

[10:00:10] first: 1
[10:00:10] first: 2
[10:00:10] first: 3
[10:00:10] first: 4
[10:00:10] first: 5
[10:00:10] second: 1
[10:00:10] second: 2

[10:00:12] second: 3
[10:00:12] second: 4

[10:00:12] second: 5        

[10:00:15] first: 6
[10:00:15] first: 7
[10:00:15] first: 8
[10:00:15] first: 9
[10:00:15] first: 10
[10:00:15] second: 6
[10:00:15] second: 7    

[10:00:17] second: 8
[10:00:17] second: 9

[10:00:19] second: 10
...