Python多进程环境下日志管理的最佳实践与实战指南

import logging

# 配置基本的日志格式
logging.basicconfig(level=logging.info, 
                    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')

# 创建一个日志记录器
logger = logging.getlogger(__name__)

# 使用日志记录器
logger.info("这是一条信息日志")
logger.warning("这是一条警告日志")
logger.error("这是一条错误日志")

import logging
import multiprocessing
import random
import time

def worker_process(queue):
    logger = logging.getlogger(f"worker-{multiprocessing.current_process().name}")
    for _ in range(5):
        time.sleep(random.random())
        logger.info(f"worker {multiprocessing.current_process().name} is working")
        queue.put(logger.name + ": " + f"worker {multiprocessing.current_process().name} is working")

def logger_process(queue):
    logger = logging.getlogger("loggerprocess")
    logger.setlevel(logging.info)
    handler = logging.filehandler("multiprocess.log")
    formatter = logging.formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
    handler.setformatter(formatter)
    logger.addhandler(handler)

    while true:
        try:
            record = queue.get()
            if record == "stop":
                break
            logger.info(record)
        except exception:
            import sys, traceback
            print('whoops! problem:', file=sys.stderr)
            traceback.print_exc(file=sys.stderr)

if __name__ == "__main__":
    queue = multiprocessing.queue(-1)
    logger_p = multiprocessing.process(target=logger_process, args=(queue,))
    logger_p.start()

    workers = []
    for i in range(5):
        worker = multiprocessing.process(target=worker_process, args=(queue,))
        workers.append(worker)
        worker.start()

    for worker in workers:
        worker.join()

    queue.put("stop")
    logger_p.join()

import multiprocessing
import logging
from logging.handlers import rotatingfilehandler
import time
import random
import os

class multiprocesssafehandler(rotatingfilehandler):
    def __init__(self, filename, mode='a', maxbytes=0, backupcount=0, encoding=none, delay=false):
        super().__init__(filename, mode, maxbytes, backupcount, encoding, delay)
        self.mode = mode
        self.encoding = encoding
        self.delay = delay
        self.maxbytes = maxbytes
        self.backupcount = backupcount

    def emit(self, record):
        try:
            if self.shouldrollover(record):
                self.dorollover()
            logging.filehandler.emit(self, record)
        except exception:
            self.handleerror(record)

    def dorollover(self):
        if self.stream:
            self.stream.close()
            self.stream = none
        if self.backupcount > 0:
            for i in range(self.backupcount - 1, 0, -1):
                sfn = self.rotation_filename("%s.%d" % (self.basefilename, i))
                dfn = self.rotation_filename("%s.%d" % (self.basefilename, i + 1))
                if os.path.exists(sfn):
                    if os.path.exists(dfn):
                        os.remove(dfn)
                    os.rename(sfn, dfn)
            dfn = self.rotation_filename(self.basefilename + ".1")
            if os.path.exists(dfn):
                os.remove(dfn)
            self.rotate(self.basefilename, dfn)
        if not self.delay:
            self.stream = self._open()

    def shouldrollover(self, record):
        if self.stream is none:
            self.stream = self._open()
        if self.maxbytes > 0:
            msg = "%s\n" % self.format(record)
            self.stream.seek(0, 2)
            if self.stream.tell() + len(msg) >= self.maxbytes:
                return 1
        return 0

def worker_process(name):
    logger = logging.getlogger(name)
    for _ in range(5):
        time.sleep(random.random())
        logger.info(f"worker {name} is working")

if __name__ == "__main__":
    log_file = "multiprocess_safe.log"
    handler = multiprocesssafehandler(log_file, maxbytes=1024, backupcount=5)
    formatter = logging.formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
    handler.setformatter(formatter)

    root_logger = logging.getlogger()
    root_logger.setlevel(logging.info)
    root_logger.addhandler(handler)

    processes = []
    for i in range(5):
        p = multiprocessing.process(target=worker_process, args=(f"worker-{i}",))
        processes.append(p)
        p.start()

    for p in processes:
        p.join()

import multiprocessing
import logging
import time
import random

def worker_process(name):
    logger = multiprocessing.get_logger()
    for _ in range(5):
        time.sleep(random.random())
        logger.info(f"worker {name} is working")

if __name__ == "__main__":
    multiprocessing.log_to_stderr(logging.info)

    processes = []
    for i in range(5):
        p = multiprocessing.process(target=worker_process, args=(f"worker-{i}",))
        processes.append(p)
        p.start()

    for p in processes:
        p.join()

import logging
import multiprocessing
from contextlib import contextmanager

@contextmanager
def log_manager(name):
    logger = logging.getlogger(name)
    handler = logging.filehandler(f"{name}.log")
    formatter = logging.formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
    handler.setformatter(formatter)
    logger.addhandler(handler)
    logger.setlevel(logging.info)
    
    try:
        yield logger
    finally:
        handler.close()
        logger.removehandler(handler)

def worker_process(name):
    with log_manager(name) as logger:
        for i in range(5):
            logger.info(f"worker {name} is working - step {i}")

if __name__ == "__main__":
    processes = []
    for i in range(5):
        p = multiprocessing.process(target=worker_process, args=(f"worker-{i}",))
        processes.append(p)
        p.start()

    for p in processes:
        p.join()

# logging.yaml 配置文件内容
"""
version: 1
formatters:
  standard:
    format: '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
handlers:
  console:
    class: logging.streamhandler
    level: debug
    formatter: standard
    stream: ext://sys.stdout
  file:
    class: logging.handlers.rotatingfilehandler
    level: info
    formatter: standard
    filename: multiprocess_app.log
    maxbytes: 10485760
    backupcount: 5
    encoding: utf8
loggers:
  worker:
    level: info
    handlers: [console, file]
    propagate: no
root:
  level: info
  handlers: [console]
"""
```python
import logging.config
import multiprocessing
import yaml
import os
def setup_logging(config_path='logging.yaml', default_level=logging.info):
    if os.path.exists(config_path):
        with open(config_path, 'rt') as f:
            try:
                config = yaml.safe_load(f.read())
                logging.config.dictconfig(config)
            except exception as e:
                print(f'error in logging configuration: {e}')
                logging.basicconfig(level=default_level)
    else:
        logging.basicconfig(level=default_level)
        print('failed to load configuration file. using default configs')
def worker_process(name):
    logger = logging.getlogger(f"worker.{name}")
    for i in range(5):
        logger.info(f"worker {name} processing task {i}")
        time.sleep(random.random())
if __name__ == "__main__":
    setup_logging()
    processes = []
    for i in range(5):
        p = multiprocessing.process(target=worker_process, args=(f"worker-{i}",))
        processes.append(p)
        p.start()
    for p in processes:
        p.join()

import logging
import multiprocessing
import time
import random

class processfilter(logging.filter):
    """自定义进程过滤器，用于过滤特定进程的日志"""
    
    def __init__(self, process_name=none):
        super().__init__()
        self.process_name = process_name

    def filter(self, record):
        if self.process_name is none:
            return true
        return record.processname == self.process_name

def setup_logger(name, log_file, level=logging.info, process_name=none):
    formatter = logging.formatter(
        '%(asctime)s - %(processname)s - %(name)s - %(levelname)s - %(message)s'
    )
    
    handler = logging.filehandler(log_file)
    handler.setformatter(formatter)
    
    logger = logging.getlogger(name)
    logger.setlevel(level)
    
    if process_name:
        process_filter = processfilter(process_name)
        handler.addfilter(process_filter)
    
    logger.addhandler(handler)
    return logger

def worker_task(name):
    logger = setup_logger(
        name=f"worker.{name}",
        log_file="filtered_processes.log",
        process_name=multiprocessing.current_process().name
    )
    
    for i in range(5):
        logger.info(f"processing task {i}")
        time.sleep(random.random())

if __name__ == "__main__":
    processes = []
    for i in range(3):
        p = multiprocessing.process(
            target=worker_task,
            name=f"worker-{i}",
            args=(f"worker-{i}",)
        )
        processes.append(p)
        p.start()

    for p in processes:
        p.join()

import logging
import multiprocessing
import queue
import threading
import time
import random
from datetime import datetime

class logaggregator:
    def __init__(self, output_file):
        self.output_file = output_file
        self.log_queue = multiprocessing.queue()
        self.should_stop = multiprocessing.event()
        self.aggregator_process = none

    def start(self):
        self.aggregator_process = multiprocessing.process(
            target=self._aggregate_logs
        )
        self.aggregator_process.start()

    def stop(self):
        self.should_stop.set()
        self.log_queue.put(none)  # 发送停止信号
        if self.aggregator_process:
            self.aggregator_process.join()

    def _aggregate_logs(self):
        with open(self.output_file, 'a') as f:
            while not self.should_stop.is_set():
                try:
                    log_entry = self.log_queue.get(timeout=1)
                    if log_entry is none:
                        break
                    f.write(f"{log_entry}\n")
                    f.flush()
                except queue.empty:
                    continue

    def log(self, message, level="info", process_name=none):
        timestamp = datetime.now().strftime('%y-%m-%d %h:%m:%s.%f')[:-3]
        process_name = process_name or multiprocessing.current_process().name
        log_entry = f"{timestamp} - {process_name} - {level} - {message}"
        self.log_queue.put(log_entry)

def worker_process(aggregator, worker_id):
    for i in range(5):
        message = f"worker {worker_id} processing task {i}"
        aggregator.log(message)
        time.sleep(random.random())

if __name__ == "__main__":
    # 创建日志聚合器
    aggregator = logaggregator("aggregated_logs.log")
    aggregator.start()

    # 创建多个工作进程
    processes = []
    for i in range(3):
        p = multiprocessing.process(
            target=worker_process,
            args=(aggregator, i)
        )
        processes.append(p)
        p.start()

    # 等待所有进程完成
    for p in processes:
        p.join()

    # 停止日志聚合器
    aggregator.stop()

import logging
import multiprocessing
import os
from datetime import datetime
import time
import random

class multilevellogger:
    def __init__(self, base_dir="logs"):
        self.base_dir = base_dir
        self.levels = {
            'debug': logging.debug,
            'info': logging.info,
            'warning': logging.warning,
            'error': logging.error,
            'critical': logging.critical
        }
        self._setup_directories()
        self._setup_loggers()

    def _setup_directories(self):
        for level in self.levels.keys():
            dir_path = os.path.join(self.base_dir, level.lower())
            os.makedirs(dir_path, exist_ok=true)

    def _setup_loggers(self):
        self.loggers = {}
        for level_name, level_value in self.levels.items():
            logger = logging.getlogger(f"multi_level.{level_name}")
            logger.setlevel(level_value)

            # 创建文件处理器
            log_file = os.path.join(
                self.base_dir,
                level_name.lower(),
                f"{level_name.lower()}_{datetime.now().strftime('%y%m%d')}.log"
            )
            handler = logging.filehandler(log_file)
            
            # 设置格式化器
            formatter = logging.formatter(
                '%(asctime)s - %(processname)s - %(name)s - %(levelname)s - %(message)s'
            )
            handler.setformatter(formatter)
            
            logger.addhandler(handler)
            self.loggers[level_name] = logger

    def log(self, level, message):
        if level in self.loggers:
            self.loggers[level].log(self.levels[level], message)

def worker_process(logger, worker_id):
    levels = ['debug', 'info', 'warning', 'error', 'critical']
    for i in range(5):
        level = random.choice(levels)
        message = f"worker {worker_id} generated {level} message for task {i}"
        logger.log(level, message)
        time.sleep(random.random())

if __name__ == "__main__":
    # 创建多级日志记录器
    multi_logger = multilevellogger()

    # 创建多个工作进程
    processes = []
    for i in range(3):
        p = multiprocessing.process(
            target=worker_process,
            args=(multi_logger, i)
        )
        processes.append(p)
        p.start()

    # 等待所有进程完成
    for p in processes:
        p.join()