Metadata-Version: 2.4
Name: py-async-log
Version: 0.0.3
Summary: A high-performance multiprocess-safe asynchronous logging tool.一个支持多进程多线程的日志封装库
Author-email: vin <techcn@qq.com>
License: Apache-2.0
Keywords: logging,multiprocessing,async,queue
Classifier: Intended Audience :: Developers
Classifier: Topic :: System :: Logging
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Operating System :: OS Independent
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Programming Language :: Python :: 3.12
Requires-Python: >=3.8
Description-Content-Type: text/markdown
License-File: LICENSE
Dynamic: license-file


# py-async-log

[![License](https://img.shields.io/badge/License-Apache%202.0-blue.svg)](https://opensource.org/licenses/Apache-2.0)
[![Python Version](https://img.shields.io/badge/python-3.8%2B-blue)](https://www.python.org/)

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

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

## 🚀 核心特性

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

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

---

## 📦 安装

使用 pip 直接安装：

```bash
pip install py-async-log
````

-----

## 🛠️ 快速开始

```python

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](LICENSE) 协议。
