Python QueueHandler: собрать логи multiprocessing

Категория: Python · Язык: Python · Обновлено:

QueueHandler отправляет сообщения из worker-процессов в очередь, а QueueListener передаёт их обработчику в родительском процессе. Так не приходится создавать файловый handler в каждом worker, а форматирование и вывод логов остаются в одном месте.

Python
import logging
import multiprocessing as mp
from logging.handlers import QueueHandler, QueueListener


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


def process_item(number):
    logging.getLogger("worker").info("processed item=%s", number)
    return number


def main():
    log_queue = mp.Queue()
    sink = logging.StreamHandler()
    sink.setFormatter(
        logging.Formatter(
            "%(asctime)s %(processName)s %(levelname)s %(message)s"
        )
    )
    listener = QueueListener(
        log_queue, sink, respect_handler_level=True
    )
    listener.start()
    try:
        with mp.Pool(
            processes=4,
            initializer=configure_worker,
            initargs=(log_queue,),
        ) as pool:
            pool.map(process_item, range(8))
    finally:
        listener.stop()
        sink.close()
        log_queue.close()
        log_queue.join_thread()


if __name__ == "__main__":
    main()

Функции worker и initializer должны быть на уровне модуля; защита if __name__ == "__main__" необходима при запуске с spawn, в том числе в Windows. QueueHandler и QueueListener есть с Python 3.2, аргумент respect_handler_level — с Python 3.5. Для multiprocessing используйте multiprocessing.Queue, а не SimpleQueue; не направляйте DEBUG-логи внутреннего multiprocessing-логгера в ту же очередь — это может привести к рекурсии или deadlock. См. logging.handlers.