带有运行状况检查的长时间运行任务答案

【问题标题】：Long running task with healthchecks带有运行状况检查的长时间运行任务
【发布时间】：2021-09-02 03:22:40
【问题描述】：

我目前正在实现一个容器化的 Python 应用程序来处理来自队列的消息。

主进程每隔 n 秒轮询一次队列，然后处理它收到的所有消息。不过，我还希望这个应用程序公开一个带有健康检查和其他端点的 API，这些端点可以将作业发送到主进程。

我想知道在 python 中执行此操作的标准库是什么，如果它们存在的话。我在 FastAPI 上看到了一些使用后台任务的示例，但这不符合我的要求，因为服务应该在启动时轮询队列，而不向其端点发出任何请求。

我也看到了提到的 Celery 库，但它似乎比我需要的复杂性大。

是否有一种简单的方法可以“并行”运行 FastAPI 应用程序和长时间运行的进程，并且两者都可以通信？

【问题讨论】：

标签： python python-multiprocessing fastapi

【解决方案1】：

在线程中启动其他任务或使用 asyncio 之后，您可以像 this 示例中那样以编程方式运行 FastAPI。这样，您应该能够从服务器端点与您之前启动的任何对象进行通信。

【讨论】：

【解决方案2】：

多处理模块有自己的队列版本。在您的调用程序中，首先创建一个队列，如下所示：

import multiprocessing as mp
self.outq_log=mp.Queue()

然后在调用 mp.Process() 以启动长时间运行的任务时，将此 Queue 对象放入参数中，从而将其传递给进程。

然后调用程序中检查队列中消息的函数如下所示：

def service_queues(self):
  #look for data from the process, print it in the windows
  try:
    if self.outq_log is not None:
      x=self.outq_log.get(block=False)
      self.logtxt.write(x)
  except queue.Empty:
    pass

最后，在您的长期运行过程中，您可以使用

将项目发送给调用者

outq_log.put(stuff)

如果您想以另一种方式从调用者向任务发送消息，您可以创建一个单独的队列并在调用者中“放置”并在任务中“获取”。

【讨论】：