Многопоточное логирование в Python: Руководство по QueueHandler и QueueListener

import logging
import logging.handlers
import queue
import concurrent.futures
import json
from contextlib import contextmanager

# ------------------------------------------------------------
# Настройка логирования с правильным разделением
# ------------------------------------------------------------
def setup_logging():
    # Создаем единую очередь для всех записей лога
    log_queue = queue.Queue(-1)  # -1 означает неограниченный размер
    
    # Создаем единственный логгер для приложения
    logger = logging.getLogger("app_logger")
    logger.setLevel(logging.DEBUG)
    
    # Добавляем QueueHandler в логгер
    queue_handler = logging.handlers.QueueHandler(log_queue)
    logger.addHandler(queue_handler)
    
    # Обработчик консоли с простым форматировщиком
    console_handler = logging.StreamHandler()
    console_handler.setFormatter(logging.Formatter(
        "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
    ))
    console_handler.setLevel(logging.INFO)
    
    # Обработчик файла с форматировщиком JSON
    json_handler = logging.FileHandler("output.log")
    json_handler.setFormatter(JsonFormatter())
    json_handler.setLevel(logging.INFO)
    
    # QueueListener с обоими обработчиками
    listener = logging.handlers.QueueListener(
        log_queue,
        console_handler,
        json_handler
    )
    
    return listener, logger

# Пользовательский форматировщик JSON
class JsonFormatter(logging.Formatter):
    def format(self, record):
        log_entry = {
            'timestamp': self.formatTime(record),
            'level': record.levelname,
            'logger': record.name,
            'message': record.getMessage(),
            'thread': record.threadName,
            'module': record.module,
            'function': record.funcName,
            'line': record.lineno
        }
        if record.exc_info:
            log_entry['exception'] = self.formatException(record.exc_info)
        return json.dumps(log_entry)

class CustomQueueListener(logging.handlers.QueueListener):
    def __init__(self, queue, *handlers, respect_handler_level=True):
        super().__init__(queue, *handlers, respect_handler_level=respect_handler_level)
    
    def handle(self, record):
        # Пользовательская логика маршрутизации записей в соответствующие обработчики
        if record.name == "json_logger":
            # Маршрутизация записей JSON-специфичных в файловый обработчик
            for handler in self.handlers:
                if isinstance(handler, logging.FileHandler):
                    handler.handle(record)
        else:
            # Маршрутизация всех остальных записей в консольный обработчик
            for handler in self.handlers:
                if isinstance(handler, logging.StreamHandler):
                    handler.handle(record)

import logging
import logging.handlers
import queue
import concurrent.futures
import json
import threading
import time
from contextlib import contextmanager

# Пользовательский форматировщик JSON
class JsonFormatter(logging.Formatter):
    def format(self, record):
        log_entry = {
            'timestamp': self.formatTime(record),
            'level': record.levelname,
            'logger': record.name,
            'message': record.getMessage(),
            'thread': record.threadName,
            'module': record.module,
            'function': record.funcName,
            'line': record.lineno
        }
        if record.exc_info:
            log_entry['exception'] = self.formatException(record.exc_info)
        return json.dumps(log_entry)

# Настройка логирования
def setup_logging():
    log_queue = queue.Queue(-1)
    
    # Основной логгер приложения
    logger = logging.getLogger("app")
    logger.setLevel(logging.DEBUG)
    
    # QueueHandler для потокобезопасности
    queue_handler = logging.handlers.QueueHandler(log_queue)
    logger.addHandler(queue_handler)
    
    # Консольный обработчик
    console_handler = logging.StreamHandler()
    console_handler.setFormatter(logging.Formatter(
        "%(asctime)s [%(threadName)s] %(levelname)s - %(message)s"
    ))
    console_handler.setLevel(logging.INFO)
    
    # Файловый обработчик с форматированием JSON
    file_handler = logging.FileHandler("output.log")
    file_handler.setFormatter(JsonFormatter())
    file_handler.setLevel(logging.DEBUG)  # Логируем все в файл
    
    # QueueListener
    listener = logging.handlers.QueueListener(
        log_queue,
        console_handler,
        file_handler
    )
    
    return listener, logger

# Функция рабочего потока
def worker(task, logger):
    try:
        logger.info(f"Начало задачи: {task}")
        
        # Имитация работы
        time.sleep(0.1)
        
        # Имитация разных уровней логирования
        if task == "error_task":
            raise ValueError("Имитируемая ошибка")
        
        result = {"task": task, "status": "completed", "timestamp": time.time()}
        logger.info(f"Результат задачи: {result}")
        
        return result
        
    except Exception as e:
        logger.error(f"Ошибка в задаче {task}: {str(e)}", exc_info=True)
        raise

# Основное выполнение
def main():
    listener, logger = setup_logging()
    listener.start()
    
    try:
        tasks = ["task1", "task2", "error_task", "task4"]
        
        with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
            futures = [
                executor.submit(worker, task, logger)
                for task in tasks
            ]
            
            # Ожидание завершения
            for future in concurrent.futures.as_completed(futures):
                try:
                    result = future.result()
                    logger.info(f"Задача успешно завершена: {result}")
                except Exception as e:
                    logger.error(f"Задача не выполнена: {str(e)}")
                    
    finally:
        listener.stop()

if __name__ == "__main__":
    main()