寻求Python中支持awaitable的进程安全Semaphore实现（适配Sanic）

直接封装multiprocessing.Semaphore为异步上下文管理器，借助线程池将同步的信号量操作转为可await的异步操作，既保证进程间安全，又不会阻塞事件循环。

实现代码 #

import asyncio
from multiprocessing import Semaphore

class AsyncProcessSemaphore:
    def __init__(self, value: int = 1, executor=None):
        # 基于multiprocessing.Semaphore实现进程间同步
        self._semaphore = Semaphore(value)
        # 可选传入自定义线程池，比如Sanic内置的executor
        self._executor = executor

    async def acquire(self) -> None:
        loop = asyncio.get_running_loop()
        # 将同步acquire放到线程池执行，避免阻塞事件循环
        await loop.run_in_executor(self._executor, self._semaphore.acquire)

    async def release(self) -> None:
        loop = asyncio.get_running_loop()
        await loop.run_in_executor(self._executor, self._semaphore.release)

    # 支持async with语法
    async def __aenter__(self):
        await self.acquire()
        return self

    async def __aexit__(self, exc_type, exc, tb):
        await self.release()

在Sanic中的使用示例 #

from sanic import Sanic, response

app = Sanic("Process_Safe_Semaphore_Demo")
# 初始化信号量，允许同时3个进程进入临界区，复用Sanic内置线程池
sem = AsyncProcessSemaphore(3, executor=app.executor)

@app.route("/critical")
async def critical_section(request):
    async with sem:
        # 这里是需要进程同步的临界区逻辑
        # 比如访问共享数据库、调用限流的外部API等
        await asyncio.sleep(1)
        return response.text(f"Process {request.app.pid} completed")

if __name__ == "__main__":
    # Sanic多进程启动，这里指定4个worker进程
    app.run(host="0.0.0.0", port=8000, workers=4)

关键说明 #

• 进程安全保障：multiprocessing.Semaphore基于系统级信号量实现，天然支持亲缘进程（Sanic的worker是通过fork主进程得到的）间的同步，能有效避免多进程竞争共享资源的问题。
• 非阻塞特性：通过loop.run_in_executor将同步的acquire/release操作放到线程池执行，当信号量需要阻塞时，事件循环会自动切换到其他任务，不会卡住整个进程的请求处理。
• 线程池复用：推荐使用Sanic内置的app.executor，避免额外创建线程池带来的资源开销。

内容的提问来源于stack exchange，提问作者freebie