如何在Python中运行后台任务

Question

我正在使用Flask开发一个小型Web服务，需要运行后台任务，最好是通过任务队列。然而，在搜索该主题后，唯一的结果似乎就是Celery和Redis Queue，它们需要单独的排队服务，因此部署起来过于繁琐。由于我只想要一个简单的后台任务队列，可以在不同的线程/进程上排队和执行任务，是否有人知道是否有类似Python的工具可用？

- RAM

---

你期望使用什么来处理队列，而不是像Redis、RabbitMQ或SQS这样的队列系统？ - jordanm

在Python3中，您可以使用线程池和进程池。也许这对您的用例已经足够了。请参阅：https://docs.python.org/3/library/concurrent.futures.html - ofirule

我强烈建议你不要在Flask控制器中使用子进程或线程调用。 - jordanm

@jordanm 为什么你不喜欢在 Flask 控制器中使用线程调用？ - RAM

由于线程代码中的错误不仅会导致当前请求失败，还可能会对运行容器（例如uwsgi、gunicorn）造成问题。uwsgi默认甚至不允许线程。这也使得代码更难测试和维护。在这种情况下，已经有一个被广泛接受的标准来实现这个目标：Celery。 - jordanm

Answer 1

import threading
import time

class BackgroundTasks(threading.Thread):
    def run(self,*args,**kwargs):
        while True:
            print('Hello')
            time.sleep(1)

t = BackgroundTasks()
t.start()

在 while 语句之后，您可以将要在后台运行的代码放置在其中。可能是删除一些模型、发送电子邮件或其他操作。

- Farhan Syedain

---

当我退出程序时，后台任务也会退出。有没有办法阻止这种情况发生？ - undefined

Answer 2

Asyncio 库可能是你正在寻找的内容。

import asyncio

async def main():
    print('Hello ...')
    await asyncio.sleep(1)
    print('... World!')

# Python 3.7+
asyncio.run(main())

- user3919706

Answer 3

如果您正在使用 FastAPI，则已经实现了BackgroundTasks。

在接收请求后执行后台任务时，不需要实现线程和队列。

代码实现：

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

# function that will run in background after sending the response
def write_notification(email: str, message=""):
    with open("log.txt", mode="w") as email_file:
        content = f"notification for {email}: {message}"
        email_file.write(content)


@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_notification, email, message="some notification")
    return {"message": "Tasks are happening in background"}

- Prakash Dahal