我心目中的想法是创建一个通用的BackgroundTask类，可以在Web服务器或独立脚本中使用，以调度不需要阻塞的任务。
我不想在这里使用任何任务队列（如celery、rabbitmq等），因为我所考虑的任务太小而且运行速度很快。只是想尽可能将它们完成并脱离掉。这是否是异步方法？将它们扔到另一个进程中？
我想出的第一个解决方案是可行的：
我认为这样做可能错过了重点。如果我将此代码与其他异步代码一起运行，那么是的，我会获得性能优势，因为阻塞操作不再阻塞主线程。
我在考虑，也许需要像单独的进程来处理这些后台任务，而完全不会阻塞主线程（上述异步代码仍将在主线程上运行）。
是否有意义创建一个处理后台作业的单独线程？像简单的作业队列，但非常轻量且不需要额外的基础设施？
还是创建类似上面的解决方案更合适？
我看到Starlette做了类似的事情（https://github.com/encode/starlette/blob/decc5279335f105837987505e3e477463a996f3e/starlette/background.py#L15），但他们在返回响应后等待后台任务。
这使得他们的解决方案依赖于Web服务器设计（即在响应后执行操作是可以的）。我想知道是否可以构建更通用的东西，使您可以在脚本或Web服务器中运行后台任务，而不会牺牲性能。
我不太熟悉异步/并发功能，因此不知道如何比较这些解决方案。看起来像是一个有趣的问题！
以下是我尝试在另一个进程上执行任务时想出的内容：
我不想在这里使用任何任务队列（如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: typing.Callable[P, typing.Any], *args: P.args, **kwargs: P.kwargs) -> None:
"""Uses singleton instance of BackgroundTask to add a task to the async execution queue.
Args:
func (typing.Callable[P, typing.Any]): _description_
"""
self.func = func
self.args = args
self.kwargs = kwargs
self.is_async = asyncio.iscoroutinefunction(func)
async def __call__(self) -> None:
if self.is_async:
with self.lock:
task = asyncio.create_task(self.func(*self.args, **self.kwargs))
self.background_tasks.add(task)
print(len(self.background_tasks))
task.add_done_callback(self.background_tasks.discard)
# TODO: Create sync task (this will follow a similar pattern)
async def create_background_task(func: typing.Callable[P, typing.Any], *args: P.args, **kwargs: P.kwargs) -> None:
b = BackgroundTask(func, *args, **kwargs)
await b()
# Usage:
async def sleep(t):
time.sleep(t)
await create_background_task(sleep, 5)
我认为这样做可能错过了重点。如果我将此代码与其他异步代码一起运行，那么是的，我会获得性能优势，因为阻塞操作不再阻塞主线程。
我在考虑，也许需要像单独的进程来处理这些后台任务，而完全不会阻塞主线程（上述异步代码仍将在主线程上运行）。
是否有意义创建一个处理后台作业的单独线程？像简单的作业队列，但非常轻量且不需要额外的基础设施？
还是创建类似上面的解决方案更合适？
我看到Starlette做了类似的事情（https://github.com/encode/starlette/blob/decc5279335f105837987505e3e477463a996f3e/starlette/background.py#L15），但他们在返回响应后等待后台任务。
这使得他们的解决方案依赖于Web服务器设计（即在响应后执行操作是可以的）。我想知道是否可以构建更通用的东西，使您可以在脚本或Web服务器中运行后台任务，而不会牺牲性能。
我不太熟悉异步/并发功能，因此不知道如何比较这些解决方案。看起来像是一个有趣的问题！
以下是我尝试在另一个进程上执行任务时想出的内容：
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)
BackgroundTask(es.transport.close) <- Probably most common use in our codebase
"""
background_tasks = set()
executor = concurrent.futures.ProcessPoolExecutor(max_workers=2)
lock = threading.Lock()
def __init__(self, func: typing.Callable[P, typing.Any], *args: P.args, **kwargs: P.kwargs) -> None:
"""Uses singleton instance of BackgroundTask to add a task to the async execution queue.
Args:
func (typing.Callable[P, typing.Any]): _description_
"""
self.func = func
self.args = args
self.kwargs = kwargs
self.is_async = asyncio.iscoroutinefunction(func)
async def __call__(self) -> None:
if self.is_async:
with self.lock:
loop = asyncio.get_running_loop()
with self.executor as pool:
result = await loop.run_in_executor(
pool, functools.partial(self.func, *self.args, **self.kwargs))