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

Question

Я возиться с многопроцессорным модулем Python.Но что-то не работает так, как я ожидал, поэтому теперь я немного запутался.

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

Чтобы проверить это, я написал следующий код:

#!/usr/bin/python

import os
import multiprocessing

# Worker function
def worker(queueA, queueB):
    while(queueA.qsize() != 0):
        item = queueA.get()
        item = "item: " + item + ". processed by worker " + str(os.getpid())
        queueB.put(item)
    return

# IPC Manager
manager = multiprocessing.Manager()
queueA = multiprocessing.Queue()
queueB = multiprocessing.Queue()

# Fill queueA with data
for i in range(0, 10):
    queueA.put("hello" + str(i+1))

# Create processes
process1 = multiprocessing.Process(target = worker, args = (queueA, queueB,))
process2 = multiprocessing.Process(target = worker, args = (queueA, queueB,))

# Call processes
process1.start()
process2.start()

# Wait for processes to stop processing
process1.join()
process2.join()

for i in range(0, queueB.qsize()):
    print queueB.get()

И это печатает следующее:

item: hello1. processed by worker 11483
item: hello3. processed by worker 11483
item: hello4. processed by worker 11483
item: hello5. processed by worker 11483
item: hello6. processed by worker 11483
item: hello7. processed by worker 11483
item: hello8. processed by worker 11483
item: hello9. processed by worker 11483
item: hello10. processed by worker 11483
item: hello2. processed by worker 11482

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

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

Answer 1

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

Я проверил это ниже (в Python3, но он должен применяться для 2.7 какну просто измените функцию print() на оператор print):

import os
import multiprocessing

# Worker function
def worker(queueA, queueB):
    for item in iter(queueA.get, 'STOP'):
        out = str(os.getpid())
        queueB.put(out)
    return

# IPC Manager
manager = multiprocessing.Manager()
queueA = multiprocessing.Queue()
queueB = multiprocessing.Queue()

# Fill queueA with data
for i in range(0, 1000):
    queueA.put("hello" + str(i+1))

# Create processes
process1 = multiprocessing.Process(target = worker, args = (queueA, queueB,))
process2 = multiprocessing.Process(target = worker, args = (queueA, queueB,))

# Call processes
process1.start()
process2.start()

queueA.put('STOP')
queueA.put('STOP')

# Wait for processes to stop processing
process1.join()
process2.join()

all = {}
for i in range(1000):
    item = queueB.get()
    if item not in all:
        all[item] = 1
    else:
        all[item] += 1
print(all)

Мой вывод (подсчет того, сколько было сделано из каждого процесса):

{'18376': 537, 
 '18377': 463}

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

Редактировать:
Другой способ подтвердить это - добавить time.sleep(3) внутрирабочая функция

def worker(queueA, queueB):
    for item in iter(queueA.get, 'STOP'):
        time.sleep(3)
        out = str(os.getpid())
        queueB.put(out)
    return

Я выполнил тест range(10), как в исходном примере, и получил:

{'18428': 5,
 '18429': 5}