一个离谱的GitHub PR

一. 前言

事情是这样子的, 在整理python loguru日志管理工具的使用记录. 专门开了个章节用于阐述在多进程下的使用, 随即扩张了下内容, 谈到了fastapi 在uvicorn下多进程(workers )的使用, 在查询资料时, 在GitHub上看到某项目上的一个离谱PR, 然后就有了文本(不吐不快).

二. 多进程

先来简单回顾一下python的多进程.

threading --- 基于线程的并行 - Python 3.13.7 文档

multiprocessing --- 基于进程的并行 - Python 3.13.7 文档

(从目录长度, 可以略窥多进程的复杂)

在: FastAPI使用基础-性能篇 | Lian, 一文中对python的并发编程做了相对细致的对比, 这里就不一一赘述了.

但是需要记住的一个典型特点: 多进程, 环境隔离问题.

2.1 Loguru多进程支持

关于在 Windows 上的多进程支持, 文档是如此描述的: 事情在Windows上有点复杂.

Things get a little more complicated on Windows. Indeed, this operating system does not support forking, so Python has to use an alternative method to create sub-processes called " spawning" . This procedure requires the whole file where the child process is created to be reloaded from scratch. This does not interoperate very well with Loguru, causing handlers to be added twice without any synchronization or, on the contrary, not being added at all (depending on add() and remove() being called inside or outside the __main__ branch). For this reason, the logger object need to be explicitly passed as an initializer argument of your child process:

让ai生成示例代码.

import multiprocessing
import time
from loguru import logger

# 配置日志: 多进程必须开启enqueue=True
logger.remove()
logger.add(
    "process.log",
    level="INFO",
    enqueue=True,  # 关键: 通过队列实现多进程安全写入
    format="{time} | 进程 {process.id} | {message}"  # 显示进程ID
)

# 进程任务: 记录日志
def process_task(task_id):
    for k in range(3):
        logger.info(f"任务 {task_id} 第 {k + 1} 次执行")
        time.sleep(0.1)  # 模拟耗时操作

if __name__ == "__main__":
    # 创建并启动多个进程
    processes = []
    for i in range(2):
        p = multiprocessing.Process(
            target=process_task,
            args=(i,)
        )
        processes.append(p)
        p.start()

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

    # 实际上这个函数是不生效的
    logger.complete()

只打印了5条日志, 实际上还会出现3, 4条的情况.

2025-09-03T11:11:39.811995+0800 | 进程 13672 | 任务 0 第 1 次执行
2025-09-03T11:11:39.819994+0800 | 进程 14004 | 任务 1 第 1 次执行
2025-09-03T11:11:39.928472+0800 | 进程 13672 | 任务 0 第 2 次执行
2025-09-03T11:11:40.037115+0800 | 进程 13672 | 任务 0 第 3 次执行
2025-09-03T11:11:40.037115+0800 | 进程 14004 | 任务 1 第 3 次执行

让各种ai反复修改上述代码, 并没能修复该问题.

不管是Gemini还是通义, 豆包, 均无法处理...给出的均是东拼西凑的不可行代码.

---

2.2 可行方案

首先尝试修改上述的代码.

其中:

def complete(self) -> AwaitableCompleter: ...

logger.complete()

这是个可当等待的函数.

import multiprocessing
import time
from loguru import logger

logger.remove()
logger.add(
    "process.log",
    level="INFO",
    enqueue=True,  # 关键: 通过队列实现多进程安全写入
    format="{time} | 进程 {process.id} | {message}"  # 显示进程ID
)
# 进程任务: 记录日志
def process_task(task_id):
    logger.info(f"任务 {task_id} 第 0 次执行")
    time.sleep(0.1)  # 模拟耗时操作

async def test_a():
    await logger.complete()

if __name__ == "__main__":

    processes = []
    for i in range(2):
        p = multiprocessing.Process(
            target=process_task,
            args=(i,)
        )
        processes.append(p)
        p.start()

    # 等待所有进程结束
    for p in processes:
        p.join()
    import asyncio
    # 确保队列中所有日志都被处理
    asyncio.run(test_a())

修改之后, 可以正常打印出所有的日志.

稍微改动一下:

import multiprocessing
import time
from loguru import logger

# 进程任务: 记录日志
def process_task(task_id):
    logger.info(f"任务 {task_id} 第 0 次执行")
    time.sleep(0.1)  # 模拟耗时操作

async def test_a():
    await logger.complete()

if __name__ == "__main__":
	# 将logger放进 if之后
    logger.remove()
    logger.add(
        "process.log",
        level="INFO",
        enqueue=True,  # 关键: 通过队列实现多进程安全写入
        format="{time} | 进程 {process.id} | {message}"  # 显示进程ID
    )
    processes = []
    for i in range(2):
        p = multiprocessing.Process(
            target=process_task,
            args=(i,)
        )
        processes.append(p)
        p.start()

    # 等待所有进程结束
    for p in processes:
        p.join()
    import asyncio
    # 确保队列中所有日志都被处理
    asyncio.run(test_a())

修改之后, 一条日志也无法打印.

---

import multiprocessing
import time
from loguru import logger

logger.remove()
logger.add(
    "process_pool.log",
    level="INFO",
    enqueue=True,  # 关键: 通过队列实现多进程安全写入
    format="{time} | 进程 {process.id} | {message}"  # 显示进程ID
)

# 进程任务: 记录日志
def process_task(task_id):
    for k in range(3):
        logger.info(f"任务 {task_id} 第 {k + 1} 次执行")
        time.sleep(0.1)
    return task_id

if __name__ == "__main__":
    with multiprocessing.Pool(1) as pool:
        arr = (pool.apply_async(process_task, (task_id,)) for task_id in range(2))
        for task in arr:
            print(task.get())

再修改成用进程池来管理执行, 可以正常打印所有日志.

import multiprocessing
from loguru import logger
import time

def process_task(_logger, task_id):
    for k in range(3):
        _logger.info(f"任务 {task_id} 第 {k + 1} 次执行")
        time.sleep(0.1)
    return task_id

if __name__ == "__main__":
    logger.remove()  # Default "sys.stderr" sink is not picklable
    logger.add("test_file.log", enqueue=True)

    processes = []
    for i in range(2):
        p = multiprocessing.Process(
            target=process_task,
            args=(logger, i,)
        )
        processes.append(p)
        p.start()

    for p in processes:
        p.join()

正常打印所有的日志, 这种方式是文档中提及的在Windows下多进程的使用示例之一, 其关键在于将logger示例作为参数传递到每个子进程中.

上述方法也是最为直观, 容易理解的, 推荐使用.

官方文档, 更多内容见: Code Snippets and Recipes for Loguru - loguru documentation

2.3 问题分析

首先来看这部分

from loguru import logger

实际上得到的是一个实例化对象

logger = _Logger(
    core=_Core(),
    exception=None,
    depth=0,
    record=False,
    lazy=False,
    colors=False,
    raw=False,
    capture=True,
    patchers=[],
    extra={},
)

在继续之前, 需要来了解一下, 开头作者说的, 这个问题在Windows下有点复杂指定是什么.

这里需要了解一下, Windows和Linux这两大主流操作系统下, 多进程的实现差异

Windows和Linux的多进程差异

Linux下面的multiprocessing基于fork, fork之后所有的本地变量都复制一份, 因此可以使用任意的全局变量;

在Windows下面, 多进程是通过启动新进程完成的, 所有的全局变量都是重新初始化的, 在运行过程中动态生成, 修改过的全局变量是不能使用的.

python: windows和linux下multiprocessing模块创建进程的区别 - 昱成 - 博客园

This does not interoperate very well with Loguru, causing handlers to be added twice without any synchronization or, on the contrary, not being added at all (depending on add() and remove() being called inside or outside the __main__ branch)

简单理解, 可以认为是初始化操作被多次执行导致的问题.

这里来看看具体的示例, Windows的多进程对于代码的影响.

创建两个文件, main.py, test_module.py

# main.py

from test_module import test
import multiprocessing

def process_task(_i):
    test.say()

if __name__ == '__main__':
    processes = []
    for i in range(2):
        p = multiprocessing.Process(
            target=process_task,
            args=(i,)
        )
        processes.append(p)
        p.start()

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

# test_module.py

class Test:
    def __init__(self):
        print("test")
        self.a = 1

    def say(self):
        print(f'hello world {self.a}')
        self.a += 1

test = Test()

创建这两个文件, 分别在Windows下和Ubuntu之下运行

# Windows之下

PS D:\Code\python_workspace\test\test_mp> python main.py
test
test
hello world 1
test
hello world 1

# ubuntu之下

lian@jarvis:~$ cd test_mp
lian@jarvis:~/test_mp$ ls
__init__.py  main.py  test_module.py
lian@jarvis:~/test_mp$ python3 main.py
test
hello world 1
hello world 1

Windows下, Test这个类实例化了三次, 而Ubuntu之下, 只执行了一次

在Windows下, 实现在Linux的相同效果.

from test_module import Test
import multiprocessing

def process_task(_test, _i):
    _test.say()

if __name__ == '__main__':
    processes = []
    test = Test() # 在主进程中实例化对象
    for i in range(2):
        p = multiprocessing.Process(
            target=process_task,
            args=(test, i,)
        )
        processes.append(p)
        p.start()

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

class Test:
    def __init__(self):
        print("test")
        self.a = 1

    def say(self):
        print(f'hello world {self.a}')
        self.a += 1

# test = Test()

上述的示例已经很清晰的表明在不同操作系统下, 使用多进程需要审慎和注意, 特别是在引入外部包来实现某些特性时, 需要仔细检查该包是否处理了在不同操作系统下的兼容问题.

2.4 FastAPI

来进一步扩展一下, 稍作修改, 改掉main.py, 在fastapi中使用

from fastapi import FastAPI
from test_module import Test
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.test = Test()
    yield

app = FastAPI(lifespan=lifespan)

uvicorn main:app --workers 4 --host 0.0.0.0 --port 8000

这里的多进程是"全局"状态的, 不是由代码本身发起的, 而是运行环境的uvicorn决定的, 这种进程间可以认为是完全"隔离"(进程间通信更为麻烦), 不管是Windows还是Linux代码实例都需要反复初始化.

From the list of deployment concepts from above, using workers would mainly help with the replication part, and a little bit with the restarts, but you still need to take care of the others:

相当于"复刻"一份.

这个机制, 意味着一些需要全局单一实例的, 如数据库, 机器学习模型等的使用将面临很多问题, 如, 多次初始化, 加载多个模型, 这意味将消耗大量的时间和占用极大的资源, 甚至导致其他更严重的问题, 不仅如此, 同时对于如何组织管理代码也是一个挑战.

2.5 GPT-SoVITS上的一个PR

现在来看看开篇所提及的一个GitHub项目的PR, 一个大型文本-语音机器学习项目:

RVC-Boss/GPT-SoVITS: 1 min voice data can also be used to train a good TTS model! (few shot voice cloning)

50k star

在检索时, 找到这样的检索内容: fastapi 多workers 并发报错 - Issue #1722 - RVC-Boss/GPT-SoVITS

顺着这个连接, 进入, 检索这个项目下的类似内容

api_v2.py中workers大于1的时候, 子进程会一直重启 - Issue #1183 - RVC-Boss/GPT-SoVITS

在这些问题中乱窜的一位"神仙", 一开始, 误以为这位大神是在解决问题, 不仅回答了别人的问题, 还顺带着提了个PR.

但是看了这个PR, 顿时惊呆...

接口增加workers参数, 支持修改多进程数量, 提高吞吐量和并发处理能力 by v3ucn - Pull Request #1724 - RVC-Boss/GPT-SoVITS

改了个worker

对, 这位"神仙"真的就改了个worker, 然后就提了PR, 幸亏这个项目的管理者不是暴躁哥, 不然怎么都要给个中指这位神仙.

这位神仙悄咪咪拿走了项目之后, 在他的仓库里, 也保持 workers = 1这个设置, 而不是改成他PR的内容.

顿时对这个"神仙"来了兴趣, 点击了它的主页,...看起来不是菜鸟啊, 怎么会给人项目提这样的PR.

好了, 开始发现这位"神仙"好玩的地方出现了, 他的仓库了塞满了从各个地方fork来的项目, 看来是玩音频相关ai的, 这不是关键, 关键在于他主页上几个Pin的项目.

主页这几个项目, 全是将别人的项目copy 过来了, 为了好看, 不是遵循惯例fork别人的项目, 而是copy.

这位神仙国内的项目还给了出处(还刻意强调了侵权的问题), 几个老外的项目, 甚至都不提项目的原出处.

顺着他的GitHub介绍, 找到他的知识星球, 知乎等账户....

刘悦的技术博客-刘悦的技术博客-知识星球

星球权益: 整合包报错一对一远程解决, 接受AI整合包定制, AI项目落地, 涵括所有系统: Windows,Mac,Linux,可以提供原创非编译代码但需要另外付费, 可以指定其他任何人的整合包, 我1比1把功能完全复刻出来, 源码交付, 专业问题解答, 给你最牛逼的体验, 爱说实话, 比那些只会套壳AI骗钱的煞笔强. god bless u

知识星球上的表述很符合这位仁兄卖课佬的特质, 狠狠的打脸自己.

刘悦简历

这位神仙自诩是前阿里巴巴的高级工程师, 简历上看似搞web的, 对这位神仙是搞python web的.

• 著有< Tornado(龙卷风)编程实战> 出版社:北京航空航天大学出版社 ISBN:9787512442436

python web 框架, 也就是和 fastapi 一路的.

Tornado is a Python web framework and asynchronous networking library, originally developed at FriendFeed. By using non-blocking network I/O, Tornado can scale to tens of thousands of open connections, making it ideal for long polling, WebSockets, and other applications that require a long-lived connection to each user.

Tornado Web Server - Tornado 6.5.2 documentation

---

离谱TM给离谱开门, 一个专门搞python web出身的阿里巴巴的高级工程师, 出书立说的技术大拿, 在他最精通的领域, 给别人的项目提了一个白痴PR.

同样的阿里系的作品:

阿里天池出版的垃圾, 随意用几篇博客内容拼成的.

一个号称拥有中国最优质数据资源的平台, 就出版这种乐色,

2.6 小结

综上, 多进程的代码的复杂度远高于其他的并发实现, 在使用上需要审慎. 就算一个包宣称多进程安全, 也需要注意其实现的方式和具体的使用场景.

Gunicorn - Python WSGI HTTP Server for UNIX

Gunicorn 'Green Unicorn' is a Python WSGI HTTP Server for UNIX. It's a pre-fork worker model. The Gunicorn server is broadly compatible with various web frameworks, simply implemented, light on server resources, and fairly speedy.

from fastapi import FastAPI
from test_module import Test

app = FastAPI()

# 不要在原文件中初始化, 也不要在lifespan中初始化
test = Test()

@app.get("/")
async def root():
    test.say()
    return {"message": "Hello World"}

(fastapi_env) lian@jarvis:~/test_mp$ gunicorn -w 4 main:app --preload
test
[2025-09-11 08:38:38 +0000] [10802] [INFO] Starting gunicorn 23.0.0
[2025-09-11 08:38:38 +0000] [10802] [INFO] Listening at: http://127.0.0.1:8000 (10802)
[2025-09-11 08:38:38 +0000] [10802] [INFO] Using worker: sync
[2025-09-11 08:38:38 +0000] [10803] [INFO] Booting worker with pid: 10803
[2025-09-11 08:38:38 +0000] [10804] [INFO] Booting worker with pid: 10804
[2025-09-11 08:38:38 +0000] [10805] [INFO] Booting worker with pid: 10805
[2025-09-11 08:38:38 +0000] [10806] [INFO] Booting worker with pid: 10806

Command line: --preload

Default: False

Load application code before the worker processes are forked.

By preloading an application you can save some RAM resources as well as speed up server boot times. Although, if you defer application loading to each worker process, you can reload your application code easily by restarting workers.

对于多个worker上启动fastapi, gunicorn提供了--preload参数, 用于解决这种多次加载的问题.

但是也需要注意其副作用.

解决Pytorch模型在Gunicorn部署无法运行或者超时问题_gunicorn 超时-CSDN博客

三. 总结

无.