我有一个FastAPI服务，它使用granian，有6 个工作进程。

我发现日志里有信息在丢失，有请求正在被发出，但轮换后的日志中却缺少这些请求的记录。

我的服务运行在服务器上的一个Docker容器中，下面我将发送日志配置、Docker CMD和 compose。

CMD ["granian", \
     "--host", "0.0.0.0", \
     "--port", "16700", \
     "--interface", "asgi", \
     "--workers", "6", \
     "--ssl-protocol-min", "tls1.2", \
     "src.my_app.main:app"]

services:
  my-app:
    build: .
    container_name: my-app
    network_mode: "host"
    restart: unless-stopped
    volumes:
      - ./logs:/src/my_app/logs

日志配置文件：

import gzip
import logging
import logging.handlers
import os
import shutil
import tempfile
from pathlib import Path
from typing import Final


class LoggerConfig:
    _LOG_FMT: Final[str] = "%(asctime)s - %(levelname)s - %(message)s"
    _COPY_BUFSIZE: Final[int] = 1024 * 1024  # 1 MiB

    @staticmethod
    def _gzip_namer(path: str) -> str:
        """Ensure rotated log files end with .gz."""
        return path if path.endswith(".gz") else f"{path}.gz"

    @classmethod
    def _gzip_rotator(cls, source: str, dest: str) -> None:
        """
        Compress `source` -> `dest` atomically, then remove `source`.

        Atomic replace avoids leaving partial .gz files on crashes.
        """
        dest_path = Path(dest)
        dest_dir = dest_path.parent if dest_path.parent != Path("") else Path(".")

        fd, tmp = tempfile.mkstemp(prefix=".logrotate-", suffix=".gz", dir=str(dest_dir))
        tmp_path = Path(tmp)

        try:
            os.close(fd)
            with open(source, "rb") as f_in, gzip.open(tmp_path, "wb", compresslevel=6) as f_out:
                shutil.copyfileobj(f_in, f_out, length=cls._COPY_BUFSIZE)

            os.replace(tmp_path, dest)  # atomic on same filesystem
            tmp_path = None  # ownership transferred to dest
            os.remove(source)
        finally:
            if tmp_path is not None:
                try:
                    tmp_path.unlink()
                except FileNotFoundError:
                    pass

    @classmethod
    def setup_logger(
        cls,
        name: str = "my_app",
        level: int = logging.INFO,
        log_file: str = "logs/my_app.log",
        backup_count: int = 365,
        when: str = "midnight",
        interval: int = 1,
    ) -> logging.Logger:
        """Setup logger with timed rotation + gzip compression."""
        logger = logging.getLogger(name)
        logger.setLevel(level)
        logger.propagate = False  # avoid duplicate logs via root/parent handlers

        if logger.handlers:  # already configured
            return logger

        formatter = logging.Formatter(cls._LOG_FMT)

        stream_handler = logging.StreamHandler()
        stream_handler.setLevel(level)
        stream_handler.setFormatter(formatter)
        logger.addHandler(stream_handler)

        log_path = Path(log_file)
        if log_path.parent != Path("."):
            log_path.parent.mkdir(parents=True, exist_ok=True)

        file_handler = logging.handlers.TimedRotatingFileHandler(
            filename=str(log_path),
            when=when,
            interval=interval,
            backupCount=backup_count,
            encoding="utf-8",
            errors="backslashreplace",
            delay=True,
        )
        file_handler.setLevel(level)
        file_handler.suffix = "%Y-%m-%d"
        file_handler.namer = cls._gzip_namer
        file_handler.rotator = cls._gzip_rotator
        file_handler.setFormatter(formatter)
        logger.addHandler(file_handler)

        return logger

我试着让AI判断可能的问题。它说问题可能出在多个工作进程写入同一个文件，或者在轮换时。

解决方案

这是一个已解决的问题，一旦你 pip install loguru。

https://pypi.org/project/loguru

异步、线程安全、多进程安全

所有添加到日志记录器的输出目标默认都是线程安全的。它们并非多进程安全，但你可以把消息排队以确保日志的完整性。若你也想进行异步日志记录，同样的思路也适用。

logger.add("somefile.log", enqueue=True)

考虑将六个工作进程（六个GIL）切换为在现代Python解释器下运行的六个线程。

带有自带功能的标准logging模块在只在单个进程中运行时，支持线程安全的日志记录。

考虑写入六个日志文件，在日志轮换时将它们合并。