[Python] 병렬 처리

🐍 본 게시글은 Python 3.11.0 환경에서 작성되었습니다!

 

여러가지 작업을 처리해야 하는 경우 병렬 처리를 통해 훨씬 빠르게 처리할 수 있다.

Multi-Processing

4개의 작업을 순차적으로 처리해야 하는 경우, 멀티 프로세싱을 통해 병렬로 4개의 작업을 동시에 처리 할 수 있다.

멀티 프로세싱은 메모리를 독립적으로 나누어 작업을 처리한다.

• 자원을 독립적으로 점유하여 처리하기 때문에 안정적이다.
• 개별적으로 자원을 점유하기 때문에 리소스 요구량이 크다.

 

한번에 처리할 수 있는 수는 CPU Core의 개수에 따라 다르다.

import time
from multiprocessing import Pool

def cpu_intensive_task(n):
    result = 0
    for i in range(n):
        result += i ** 2
    return result

if __name__ == "__main__":
    start_time = time.time()
    results = []
    for _ in range(4):  # 4개의 작업을 순차적으로 수행
        result = cpu_intensive_task(10**6)
        results.append(result)
    end_time = time.time()

    print("Results:", results)
    print("Sigle Processing:", end_time - start_time)


    start_time = time.time()
    with Pool(processes=4) as pool:  # 4개의 프로세스를 생성
        results = pool.map(cpu_intensive_task, [10**6] * 4)  # 4개의 작업을 병렬로 수행
    end_time = time.time()

    print("Results:", results)
    print("Multi processing:", end_time - start_time)

 

 

Multi-Threading

I/O 작업을 여러개 해야할 때는 멀티 쓰레딩을 사용하는 것이 좋다.

멀티 쓰레딩은 하나의 메모리를 작업을 작게 나눠서 병렬로 처리한다.

• 자원을 공유(하나의 메모리)하기 때문에 하나의 작업에서 오류가 발생하면 다른 작업도 영향을 받는다. 

 

import time
import threading

def io_intensive_task(n):
    time.sleep(n)  # I/O 작업을 시뮬레이션 (n초 대기)
    print(f"Task {n} completed")

if __name__ == "__main__":
    start_time = time.time()

    for i in range(1, 5):  # 1초, 2초, 3초, 4초 대기 작업
        io_intensive_task(i)
    end_time = time.time()
    print("Single-Threading:", end_time - start_time)


    start_time = time.time()
    threads = []
    for i in range(1, 5):  # 1초, 2초, 3초, 4초 대기 작업
        thread = threading.Thread(target=io_intensive_task, args=(i,))
        threads.append(thread)
        thread.start()  # 스레드 시작

    for thread in threads:
        thread.join()  # 모든 스레드가 완료될 때까지 대기

    end_time = time.time()
    print("Multi-Threading:", end_time - start_time)

 

 

이 두개의 병렬 처리를 적절히 조합하면 훨씬 빠른 시간에 작업을 처리할 수 있다.

import time
import multiprocessing
import threading

def io_intensive_task(n):
    time.sleep(n)  # I/O 작업을 시뮬레이션 (n초 대기)
    print(f"Thread Task {n} completed")

def process_function(num_threads):
    threads = []
    for i in range(1, num_threads + 1):
        thread = threading.Thread(target=io_intensive_task, args=(i,))
        threads.append(thread)
        thread.start()  # 스레드 시작

    for thread in threads:
        thread.join()  # 모든 스레드가 완료될 때까지 대기

if __name__ == "__main__":
    start_time = time.time()
    num_tasks = 6  # 총 작업 수

    for i in range(1, num_tasks + 1):
        io_intensive_task(i)  # 각 작업 실행

    end_time = time.time()
    print(f"Default: {end_time - start_time:.4f}")

    start_time = time.time()
    num_processes = 2  # 프로세스 수
    num_threads_per_process = 3  # 각 프로세스에서 실행할 스레드 수

    processes = []
    for _ in range(num_processes):
        process = multiprocessing.Process(target=process_function, args=(num_threads_per_process,))
        processes.append(process)
        process.start()  # 프로세스 시작

    for process in processes:
        process.join()  # 모든 프로세스가 완료될 때까지 대기

    end_time = time.time()
    print(f"multiprocessing and multithreading:{end_time - start_time:.4f}")