многопроцессорность в Python 3.6

Question

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

Существует много информации о Process / Threads / Pools, но я не уверен, как бы я поступил в этом направлении.

Мне бы хотелось иметь возможность указывать, например, максимум 5 потоков / процессов одновременно, но я не уверен, как работает итерация (т. Е. Сначала я перебираю «задачи», или я перебираю количество потоки / процессы?

например. У меня есть один словарь, содержащий 2 уровня иерархии:

dict = {'router1': {'id': 1, 'name': 'rtr1_core'}, 'router2': {'id': 2, 'name': 'rt2_core'}, 'router3': {'id': 3, 'name': 'rtr3_access'}}

(Обратите внимание, я упростил список элементов в каждом из маршрутизаторов - реально в каждом из них есть более 20 ключей, но я использую id и name, чтобы упростить его.)

Итак, я выполняю итерацию через router1 ... router 3 и каждый словарь '{' id ': 1,' name ':' rtr1_core '}' необходимо передать в функцию 'process_routers', содержащую один аргумент (dict ).

Мне не нужно делиться информацией между процессами.

Мне удалось получить:

    # Process the data
    p = multiprocessing.Process(target=process_rtr, args=(rtr_data,))
    p.start()
    p.join()

Что, кажется, все еще запускает его в последовательном режиме - как я могу запустить их параллельно в до x потоков?

На самом деле я рассчитываю выполнить функцию process_rtr около 50 тыс. Раз в 5-10 потоках на основе нашего текущего оборудования. Поэтому было бы здорово указать количество потоков / процессов, чтобы я мог настроить это по мере необходимости.

Большое спасибо за ваше время и помощь заранее.

Фрэнк

Answer 1

Поскольку вы используете Python 3.6, рассматривали ли вы библиотеку asyncio? Я не уверен, что вы делаете в своей функции process_rtr, поэтому вам может понадобиться взглянуть на некоторые из асинхронных совместимых библиотек ( асинхронные библиотеки ). Должно быть в состоянии сделать что-то вроде этого:

import asyncio
import random
dict = {'router1': {'id': 1, 'name': 'rtr1_core'},
        'router2': {'id': 2, 'name': 'rt2_core'},
        'router3': {'id': 3, 'name': 'rtr3_access'}}

async def process_rtr(id, name):
    """Do your execution here."""
    s_time = await asyncio.sleep(random.randint(0, 5))
    print(f"Processing {id}, {name}")

loop = asyncio.get_event_loop()
tasks = [asyncio.ensure_future(process_rtr(**router_details))
                               for router, router_details
                               in dict.items()]
loop.run_until_complete(asyncio.wait(tasks))
loop.close()

Answer 2

Ваша структура немного не в порядке. Убедитесь, что вы настроили N заданий.

jobs = [Process(target=yourfunction, args=(yourdata) for i in range(numberofthreads)]

for j in jobs:
    j.start()
for j in jobs:
    j.join()

Но учтите, что если вы скажете следующее, вы можете быть разочарованы:

manager = Manager()
yourdata= manager.dict(routerdict)

Управляемый словарь соберет все ваши данные и сделает их безопасными для блокировки. Это требует времени. Дела идут намного быстрее, если у вас есть только управляемый список с идентификаторами или чем-то еще. Если yourfunction сможет получить доступ к базовым данным, извлекая их из какого-либо другого источника, вы сможете получить большую скорость. Все, что вам нужно, это управляемый / заблокированный список простых элементов (routerid1, routerid2, ...) вместо гигантского общего пакета данных (данных, которые были настолько массивны, что вы хотели распараллелить их обработку ...)