Блог

Главная - Вопросы по программированию - Сравнение многопроцессорности Python и многопоточности в файлах 100K

Сравнение многопроцессорности Python и многопоточности в файлах 100K

#python #python-3.x #multithreading #multiprocessing

#python #python-3.x #многопоточность #многопроцессорность

Вопрос:

У меня есть папка с 100 тысячами файлов общим объемом 50 ГБ. Цель состоит в том, чтобы прочитать каждый файл, запустить некоторое регулярное выражение для хранения данных. Я пытаюсь запустить тест, чтобы увидеть, какой подход, многопоточный или многопроцессорный, будет наиболее идеальным.

Сервер, который я использую, имеет 4 ядра и 8 ГБ оперативной памяти. Без какой-либо многопоточности для выполнения задачи требуется около 5 минут.

 from concurrent.futures import ThreadPoolExecutor
threads= []

def read_files(filename):
     with open(filename, 'r') as f:
             text = f.read()

with ThreadPoolExecutor(max_workers=50) as executor:
     for filename in glob.iglob('/root/my_app/my_app_venv/raw_files/*.txt', recursive=True):            
         threads.append(executor.submit(read_files, filename))

Многопоточность в среднем составляет 1 минуту и 30 секунд.

Теперь я пытаюсь настроить тест для многопроцессорности и использовать 4 ядра на сервере, но ничего не получается.

 from multiprocessing import Lock, Process, Queue, current_process
import time
import queue 


def read_files(tasks_to_accomplish):
    while True:
        try:            
            filename = tasks_to_accomplish.get_nowait()
            with open(filename, 'r') as f:
                text = f.read()
        except queue.Empty:
            break    


def main():    
    number_of_processes = 4
    tasks_to_accomplish = Queue()    
    processes = []

    for filename in glob.iglob('/root/my_app/my_app_venv/raw_files/*.txt', recursive=True):        
        tasks_to_accomplish.put(filename)
        
    # creating processes
    for w in range(number_of_processes):
        p = Process(target=read_files, args=(tasks_to_accomplish,))
        processes.append(p)
        p.start()

    # completing process
    for p in processes:
        p.join()

   
if __name__ == '__main__':
    main()

Пожалуйста, помогите!

Ответ №1:

Поскольку вы уже используете concurrent.futures , я рекомендую использовать ProcessPoolExecutor , который находится поверх multiprocessing аналогично тому, как ThreadPoolExecutor находится поверх threading . Эти классы имеют почти идентичные API

https://docs.python.org/3/library/concurrent.futures.html#processpoolexecutor