py-async-log 0.0.5

A high-performance multiprocess-safe asynchronous logging tool.一个支持多进程多线程的日志封装库

py-async-log

py-async-log 是一个高性能、多线程、多进程安全的异步日志工具，专为解决Python多线程、多进程环境下的日志竞争与阻塞问题而设计。

注意：当前是第一个版本，可能会有很多未完善的地方。

🚀 核心特性

• 多进程安全：基于 multiprocessing.Manager 队列，完美支持跨进程日志收集。
• 异步非阻塞：主进程与子进程仅负责将日志推入队列，磁盘IO由独立的监听线程处理。
• 防止乱序：确保多进程并发写入时日志记录的完整性与顺序性。
• 轻量易用：模块化单例设计，几行代码即可完成全项目日志配置。

在标准 logging 库中，多进程同时写入同一个文件会导致：

1. 内容覆盖：多个进程竞争文件锁，可能导致日志行破碎或丢失。
2. 性能瓶颈：磁盘 IO 是慢速操作，同步写入会阻塞业务逻辑。 py-async-log 通过生产者-消费者模型，将所有进程的日志汇聚到统一队列，由主进程单线程写入，彻底规避了上述问题。

---

📦 安装

使用 pip 直接安装：

pip install py-async-log

---

🛠️ 快速开始

import logging
import multiprocessing
import os
import time

from py_async_log import get_manager

# 定义一个临时的日志文件名
TEST_LOG_FILE = "test_run.log"


def worker_task(config_info, message):
    """模拟子进程任务"""
    # 解包：config_info 是 (函数, 参数)
    config_fn, args = config_info
    config_fn(*args)  # 执行初始化

    logger = logging.getLogger("test_worker")
    logger.info(message)


def test_multiprocessing_logging():
    """测试多进程日志是否能正确写入同一个文件"""

    # 1. 环境清理：如果上次测试残留了日志文件，先删除
    if os.path.exists(TEST_LOG_FILE):
        os.remove(TEST_LOG_FILE)

    # 2. 初始化管理器
    log_mgr = get_manager(TEST_LOG_FILE, level=logging.INFO)
    log_mgr.start_logging()

    try:
        # 获取可以序列化的配置信息
        config_info = log_mgr.get_worker_config_args()

        # 3. 启动多个子进程同时写日志
        messages = [f"测试消息 from process {i}" for i in range(3)]
        processes = []

        for msg in messages:
            p = multiprocessing.Process(target=worker_task, args=(config_info, msg))
            p.start()
            processes.append(p)

        for p in processes:
            p.join()

        # 4. 给一点缓冲时间让 QueueListener 完成写入操作
        time.sleep(0.5)
        log_mgr.stop_logging()

        # 5. 验证结果
        assert os.path.exists(TEST_LOG_FILE), "日志文件未生成"

        with open(TEST_LOG_FILE, "r", encoding="utf-8") as f:
            content = f.read()
            for msg in messages:
                assert msg in content, f"日志中缺失消息: {msg}"
    except Exception as e:
        print(f"测试过程中发生错误: {e}")
        raise
    finally:
        # 6. 测试完成后清理
        if os.path.exists(TEST_LOG_FILE):
            os.remove(TEST_LOG_FILE)


if __name__ == "__main__":
    test_multiprocessing_logging()

---

📜 开源协议

本项目采用 Apache License 2.0 协议。