Вопрос:
Я запускаю сценарий, чтобы взять список таблиц базы данных, проверить количество строк каждой таблицы и добавить результаты каждого запроса в словарь. Я использую многопроцессорную обработку, чтобы ускорить ее: менеджер для создания разделяемого списка и расширяемого словаря, который процессы могут читать и добавлять, и обрабатывать для создания процессов.
from multiprocessing import Process, Manager def main(): mgr = Manager() # Function to get the list of tables table_list = mgr.list(get_table_list()) counts = mgr.dict() for table in table_list: # get_table_count runs a ‘SELECT COUNT(*) FROM <table>’ and appends # the result to the counts dict p = Process(target=select_star, args=(table, counts, ‘prod’)) p.start() p.join()
Мой вопрос в том, как Process управляет количеством открытых процессов? Я привык использовать Pool() с методом apply_async(), где вы указываете количество процессов при создании пула.
Также будет оценен любой дополнительный комментарий, если это лучший метод для использования в этом приложении.
–
Благодаря dano, помогающему реализовать два решения ниже:
1 – используя Pool.apply_async, как я привык:
from multiprocessing import Process, Manager, cpu_count def main(): mgr = Manager() table_list = get_table_list() pool = Pool(cpu_count() / 2) prod_counts = mgr.dict() for table in table_list: pool.apply_async(get_table_count, args=(table, prod_counts, ‘prod’)) pool.close() pool.join()
2 – с помощью Pool.map() с itertools.partial()
from multiprocessing import Process, Manager, cpu_count def main(): mgr = Manager() table_list = get_table_list() pool = Pool(cpu_count() / 2) prod_counts = mgr.dict() func = partial(get_table_count, result_dict=prod_counts, env=’prod’) pool.map(func, table_list) print prod_counts Лучший ответ:
multiprocessing.Process не знает, сколько других процессов открыто, или что-то сделать для управления количеством запущенных объектов Process. Вам нужно использовать multiprocessing.Pool чтобы получить эту функциональность.
Когда вы используете Process напрямую, вы запускаете подпроцесс, как только вы вызываете p.start(), и ждите завершения Process при вызове p.join(). Таким образом, в вашем примере кода вы выполняете только один процесс за раз, но вы запускаете len(table_list) разные процессы.
Это не очень хороший подход; потому что вы только запускаете один процесс за раз, вы на самом деле ничего не делаете одновременно. Это будет медленнее, чем обычный однопоточный/процессный подход из-за накладных расходов на запуск подпроцесса и доступа к Manager.dict. Вместо этого вы должны просто использовать Pool:
from functools import partial from multiprocessing import Manager, Pool def select_star(table, counts, type_): # counts and type_ will always be the counts dict and «prod», respectively pass def main(): mgr = Manager() counts = mgr.dict() p = Pool() func = partial(select_star, counts, «prod») # Using a partial lets us pass extra parameters to select_start p.map(func, get_table_list()) # No need to use a manager for the list, since you’re not passing the whole thing to the children. if __name__ == «__main__»: main() Ответ №1
Из документов:
При многопроцессорной обработке процессы порождаются созданием объекта Process и последующим вызовом метода start().
Короче говоря, он не управляет количеством открытых процессов. Он просто запускает процесс, когда вы вызываете start().
