Python 다중 처리 로깅

Python의 multiprocessing 모듈은 별도의 프로세스에서 작업을 실행합니다. 개별 프로세스를 시작하고 종료될 때까지 기다리거나, Pool을 사용해 하나의 함수를 여러 입력에 적용하거나, 작업 프로세스의 로그 레코드를 부모 프로세스의 리스너로 보낼 수 있습니다. 아래 예제에서는 각 방식을 설정하는 방법과 메시지를 생성한 프로세스를 식별하는 방법을 보여 줍니다.

다중 처리

다중 처리는 서로 분리된 메모리 공간을 사용하는 별도의 프로세스에서 작업을 실행합니다. 독립된 작업으로 나눌 수 있는 CPU 중심 작업에 유용합니다. 프로세스의 실행은 운영 체제가 스케줄링하므로 출력이 표시되는 순서는 달라질 수 있습니다.

Python의 다중 처리

프로세스를 만들려면 multiprocessing 모듈을 가져옵니다. Process에는 호출 가능한 객체를 target으로, 인수 튜플을 args로 전달합니다. start()를 호출하면 작업 프로세스가 시작되고, join()을 호출하면 해당 프로세스가 끝날 때까지 기다립니다. 메인 가드는 자식 프로세스가 프로그램 최상위의 프로세스 생성 코드를 다시 실행하지 않도록 합니다. 작업 프로세스가 메인 모듈을 가져오는 방식으로 시작되는 플랫폼에서도 이 가드가 필요합니다.

다음 예제를 실행해 보세요:

import multiprocessing


def report(value):
    print(f"worker received: {value}", flush=True)


if __name__ == "__main__":
    process = multiprocessing.Process(target=report, args=(5,))
    process.start()
    process.join()
    print("parent: process finished")

worker received: 5
parent: process finished

작업 프로세스는 join()이 반환되기 전에 메시지를 출력하므로, 그 뒤에 부모 프로세스의 메시지가 표시됩니다. 작업 프로세스가 여러 개라면 메시지가 섞이거나 다른 순서로 나타날 수 있습니다. 출력을 한곳에서 처리해야 한다면 큐를 사용하세요.

프로세스 ID

작업 프로세스 안에서 os.getpid()를 사용하면 실행 중인 프로세스를 식별할 수 있습니다. start()를 호출한 뒤에는 부모 프로세스에서도 Process 객체의 pid 속성으로 같은 프로세스 ID를 확인할 수 있습니다. PID는 운영 체제가 할당하며 실행할 때마다 달라지므로 고정된 출력값이 아니라 실행 시점의 값으로 다루세요.

import multiprocessing
import os


def report_pid():
    print(f"worker PID: {os.getpid()}", flush=True)


if __name__ == "__main__":
    process = multiprocessing.Process(target=report_pid)
    process.start()
    print(f"Process.pid: {process.pid}")
    process.join()

Pool로 작업 매핑하기

Pool은 작업 프로세스 그룹을 관리합니다. map() 메서드는 반복 가능한 객체의 각 항목에 하나의 함수를 적용하고, 결과를 입력 순서대로 반환합니다. 풀 크기는 사용할 수 있는 작업 프로세스의 최대 수이며, 입력마다 프로세스 하나가 만들어진다는 뜻은 아닙니다. 컨텍스트 관리자를 빠져나오면 블록이 끝날 때 풀이 강제 종료됩니다. 정상적인 완료가 중요하다면, 대신 풀을 새 작업에 대해 명시적으로 닫고 join()으로 작업 프로세스가 끝나기를 기다리세요.

import multiprocessing


def square(value):
    return value * value


if __name__ == "__main__":
    with multiprocessing.Pool(processes=2) as pool:
        results = pool.map(square, [2, 3, 4])
    print(results)

Python의 다중 처리 로깅

작업 프로세스는 로그 레코드를 하나의 큐로 보내고, 부모 프로세스의 QueueListener가 해당 레코드를 출력 핸들러에 전달하도록 구성합니다. 이렇게 하면 파일이나 콘솔 출력을 한 프로세스에서 처리하므로 여러 작업 프로세스가 같은 핸들러에 각각 쓰는 일을 피할 수 있습니다. 각 작업 프로세스의 로거에는 큐 핸들러를 설정하고, 리스너에는 큐와 하나 이상의 핸들러를 전달합니다.

다음 전체 예제는 풀의 작업 프로세스에 QueueHandler를 설정합니다. 부모 프로세스는 풀을 만들기 전에 리스너를 시작합니다. 매핑이 끝나면 풀에 새 작업이 들어오지 않도록 닫고, 작업 프로세스들이 끝날 때까지 기다린 다음 리스너를 중지합니다. 작업 프로세스가 종료될 때 각 큐 공급기에 버퍼링된 레코드를 플러시하므로 리스너가 남은 레코드를 처리할 수 있습니다. 매핑 중 예외가 발생해도 같은 종료 절차가 실행되므로, 예외가 전파되기 전에 완료된 작업의 레코드를 처리할 수 있습니다. 포매터는 각 메시지에 로그 수준을 추가하고, 부모 프로세스는 레코드를 stdout에 씁니다. 작업 프로세스 하나가 항목을 순서대로 처리합니다. 프로세스 ID로 작업 프로세스를 식별하려면 앞의 PID 예제를 사용하세요.

import logging
import multiprocessing
import sys
from logging.handlers import QueueHandler, QueueListener


def configure_worker(log_queue):
    logger = logging.getLogger()
    logger.handlers.clear()
    logger.setLevel(logging.INFO)
    logger.addHandler(QueueHandler(log_queue))


def report(value):
    logger = logging.getLogger(__name__)
    logger.info("item=%s", value)
    return value


if __name__ == "__main__":
    log_queue = multiprocessing.Queue()
    output_handler = logging.StreamHandler(sys.stdout)
    output_handler.setFormatter(logging.Formatter("%(levelname)s %(message)s"))
    listener = QueueListener(log_queue, output_handler)
    listener.start()

    try:
        pool = multiprocessing.Pool(
            processes=1,
            initializer=configure_worker,
            initargs=(log_queue,),
        )
        try:
            results = pool.map(report, [1, 2, 3])
        finally:
            pool.close()
            pool.join()
    finally:
        listener.stop()
        log_queue.close()
        log_queue.join_thread()

INFO item=1
INFO item=2
INFO item=3

이 예제는 종료 코드 0으로 완료되었으며 매핑된 각 항목에 대해 INFO 레코드를 하나씩 출력했습니다. 작업 프로세스가 하나였으므로 레코드는 매핑 순서대로 나타났습니다.

콘솔 대신 파일에 레코드를 쓰려면 logging.StreamHandler(sys.stdout)을 logging.FileHandler("multiprocessing.log", encoding="utf-8")로 바꾸세요. 해당 핸들러의 소유자와 종료 책임은 부모 프로세스에 있습니다. 인터프리터 종료 시 정리에 맡기지 않고 직접 관리한다면 listener.stop()을 호출한 뒤 핸들러를 닫으세요. 로그를 순환 저장하거나 여러 대상으로 보내려면 추가 핸들러를 QueueListener에 전달하고 각 핸들러에 필요한 포매터를 설정하면 됩니다.