ConcurrentLogHandler不能直接用于asyncio环境。因其依赖os.fork()和信号处理，且日志写入同步阻塞，即使配合to_thread也无法根本解决事件循环抢占和并发安全问题。
ConcurrentLogHandler 能不能直接用在 asyncio 环境里？
不能。它内部依赖 os.fork() 和信号处理（如 SIGHUP），在 asyncio 的单线程事件循环中既不安全，也无法响应异步 I/O。更关键的是：它的日志写入是同步阻塞的——哪怕你用 asyncio.to_thread() 包一层，也只是把阻塞操作扔到线程池，没解决根本问题：日志调用仍会抢占事件循环资源，高并发下容易拖慢整个服务。
为什么不用 logging.handlers.RotatingFileHandler + asyncio.to_thread？
可以临时用，但有隐患：
-
RotatingFileHandler的doRollover()是原子性极弱的操作：先重命名旧文件，再新建文件，中间若被中断（比如异步任务被取消、进程崩溃），极易产生损坏或丢失日志 - 多个协程同时触发滚动时，
to_thread无法保证互斥，可能两个线程同时执行os.rename()，导致 OSError: [Errno 17] File exists 或更糟的覆盖 - 它不感知 asyncio 生命周期，无法在
asyncio.run()结束前确保缓冲日志刷盘，容易丢最后几条
真正可行的异步日志滚动方案：用 aiofiles + 自定义 Handler
核心思路是绕过标准库的同步 handler，自己实现一个基于 aiofiles 的异步写入器，并把滚动逻辑完全控制在协程内。关键点：
macOS UI 自动化技能包 —— 用 Python 控制你的 Mac
下载
让 macOS 自动化像说话一样简单 —— 用代码控制你的 Mac，无需学习 AppleScript 系统要求 macOS 14+ (Sonoma) Python 3.9+ 需要开启：辅助功能 + 屏幕录制权限
- 滚动判断必须在协程中完成（比如检查文件大小），不能依赖
RotatingFileHandler的shouldRollover()同步钩子 - 滚动动作本身要用
await aiofiles.os.replace()（注意不是os.replace）并加锁，避免竞态 - 所有写入必须走
await f.write()，且缓冲区要手动管理（aiofiles默认无行缓冲，需显式await f.flush()） - 示例片段：
import aiofiles
import asyncio
import os
<p>class AsyncRotatingFileHandler:
def <strong>init</strong>(self, filename, max_bytes=10_485_760, backup_count=5):
self.filename = filename
self.max_bytes = max_bytes
self.backup_count = backup_count
self._lock = asyncio.Lock()</p><pre class="brush:python;toolbar:false;">async def _get_file_size(self):
try:
return (await aiofiles.os.stat(self.filename)).st_size
except FileNotFoundError:
return 0
async def _do_rollover(self):
async with self._lock:
for i in range(self.backup_count - 1, 0, -1):
sfn = f"{self.filename}.{i}"
dfn = f"{self.filename}.{i+1}"
if await aiofiles.os.path.exists(sfn):
await aiofiles.os.replace(sfn, dfn)
if await aiofiles.os.path.exists(self.filename):
await aiofiles.os.replace(self.filename, f"{self.filename}.1")
async def emit(self, record):
if await self._get_file_size() >= self.max_bytes:
await self._do_rollover()
async with aiofiles.open(self.filename, "a") as f:
await f.write(record + "\n")
await f.flush()
ConcurrentLogHandler 唯一能“凑合”的场景
仅限于多进程模型（如 Gunicorn + sync workers）中，且主进程不跑 asyncio——此时它靠 fork + 文件锁工作正常。一旦你启用了 uvloop、用 asyncio.start_server() 写网络服务，或者混用 asyncio.create_subprocess_exec()，就别碰它。它的 C 扩展和信号机制与 event loop 天然冲突，调试时看到 RuntimeError: cannot schedule new futures after shutdown 或静默丢日志，基本就是它在作祟。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
