Понимание работы пула процессов в бесконечном цикле - PullRequest
Новая
       11

Понимание работы пула процессов в бесконечном цикле

0 голосов
/ 23 октября 2019

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

p = Pool(processes = 2)

while True:
  message = receive_message_from_queue()
  if message is not None:
      # Do some task
      p.map_async(action_fn, [temp_iterables])

Что будет здесь, если в очереди будет 100 сообщений? Будет ли 100 процессов, созданных Python? Или в любое время будут обрабатываться только два сообщения? Кроме того, в таком случае, как это можно убить процесс, когда его задача выполнена, и воссоздать процесс, когда есть новое сообщение?

1 Ответ

0 голосов
/ 23 октября 2019

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

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

Вместо этого бизнес-логика относится к компонентам, которые выполняют «реальную работу», например, обогащают или преобразуют данные, генерируют статистику и т. Д.

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

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

Задания могут планироваться неблокирующим (асинхронным) способом, позволяющим службе продолжать свой поток выполнения или в блокирующем (синхронном) режиме, останавливая выполнение дорезультаты не готовы.

В вашем конкретном примере вы объявляете пул с двумя работниками. Предполагая, что вы используете класс multiprocessing.Pool, интерпретатор запустит 2 процесса, которые будут ожидать новых заданий. Когда вы вызываете map_async, итерация разделяется на несколько кусков, которые ставятся в очередь во внутренней очереди пула. Рабочие будут выбирать фрагменты в порядке их поступления, запускать для них функцию action_fn и публиковать результаты во второй очереди результатов, которая будет использована службой.

Несколько вызовов на map_async приводят к добавлению большего количества патчей во внутреннюю очередь. Практически очередь бесконечна по размеру. Практически, если вам удастся заполнить его, последующий вызов map_async будет блокироваться, пока рабочие не освободят место для новых заданий, которые будут поставлены в очередь.

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

...