小编Ram*_*war的帖子

有谁知道为什么设计决定将“[”和“]”或“{”和“}”ASCII键码分开两个而不是一个数字？强迫症被触发。

如果我有下面的代码，变量将如何service影响端点的异步性质？变量会被共享吗？或者它会在使用时被锁定，从而阻止其他端点访问它，直到当前端点完成为止？

我问上面的问题是假设服务实例是无状态的，即如果我在每个端点中创建一个服务实例，那么这将是等效的。我不愿意这样做，因为我不知道实例化和销毁服务对象和共享服务对象哪个更耗时？

from typing import List, Union
from fastapi import APIRouter, Body, Depends

# Service definition
router = APIRouter()
service = Service()

@router.post("/a", response_model=Union[A, None])
async def x():
    service.start()
    pass

@router.post("/b", response_model=Union[B, None])
async def y():
    service.stop()
    pass

我想到的是一个非常通用的BackgroundTask 类，可以在网络服务器或独立脚本中使用，以安排不需要阻塞的任务。

我不想在这里使用任何任务队列（celery、rabbitmq 等），因为我正在考虑的任务太小且运行速度太快。只是想尽可能地完成它们。这是一种异步方法吗？将它们扔到另一个进程上？

我想出的第一个可行的解决方案：

# Need ParamSpec to get correct type hints in BackgroundTask init
P = ParamSpec("P")


class BackgroundTask(metaclass=ThreadSafeSingleton):
    """Easy way to create a background task that is not dependent on any webserver internals.

    Usage:
        async def sleep(t):
            time.sleep(t)

        BackgroundTask(sleep, 10) <- Creates async task and executes it separately (nonblocking, works with coroutines)
        BackgroundTask(time.sleep, 9) <- Creates async task and executes it separately (nonblocking, works with normal functions)
    """

    background_tasks = set()
    lock = threading.Lock()

    def __init__(self, func: …