13

我正在使用 python3 + falcon 组合编写一个 API。

在方法中有很多地方我可以向客户端发送回复，但是由于一些执行 DB、i/o 操作等的繁重代码，它必须等到繁重的部分结束。

例如：

class APIHandler:                                                                      
  def on_get(self, req, resp):
    response = "Hello"
    #Some heavy code
    resp.body(response)

我可以在第一行代码发送“Hello”。我想要的是在后台运行繁重的代码并发送响应，无论繁重的部分何时完成。

Falcon 没有任何内置的异步功能，但他们提到它可以与 gevent 之类的东西一起使用。我还没有找到任何关于如何将这两者结合起来的文档。

4

4 回答 4

8

客户端库对异步操作有不同的支持，因此决定通常归结为您的特定后端客户端最支持哪种异步方法，以及您想使用哪个 WSGI 服务器。另请参阅下面的一些更常见的选项...

对于不支持异步交互模型的库，无论是原生的还是通过某种子类化机制，任务都可以委托给线程池。对于特别长时间运行的任务（即几秒或几分钟），Celery 是一个不错的选择。

WSGI（和Falcon）应用程序的一些更常见的异步选项的简要调查：

  • 扭曲。偏爱显式异步风格，可能是最成熟的选择。为了与 Falcon 等 WSGI 框架集成，有twisted.web.wsgi和crochet。
  • 异步。借鉴了 Twisted 的许多想法，但利用 Python 3 语言特性来提供更简洁的界面。从长远来看，这可能是最干净的选择，但需要改进 WSGI 接口（另请参阅pulsar 对 PEP-3333 的扩展作为一种可能的方法）。在撰写本文时，asyncio 生态系统还比较年轻。社区仍在尝试围绕界面、模式和工具的各种方法。
  • 小事件。支持一种旨在使异步代码看起来同步的隐式样式。eventlet 执行此操作的一种方法是通过猴子修补标准库中的 I/O 模块。有些人不喜欢这种方法，因为它掩盖了异步机制，使边缘情况更难调试。
  • 事件。类似于 eventlet，虽然更现代一点。uWSGI 和 Gunicorn 都支持对标准库进行猴子补丁的 gevent worker 类型。

最后，可以将 Falcon 扩展为原生支持 twisted.web 或 asyncio (ala aiohttp)，但我认为还没有人尝试过。

于 2015-06-11T20:37:39.847 回答
5

我将 Celery 用于异步相关工作。我不知道 gevent 。看看这个http://celery.readthedocs.org/en/latest/getting-started/introduction.html

于 2015-02-10T09:03:26.823 回答
3

我认为这里有两种不同的方法：

  1. 任务管理器（如 Celery）
  2. 异步实现（如 gevent）

你用他们每个人取得的成就是不同的。使用 Celery，您可以做的是运行同步计算响应所需的所有代码，然后在后台运行任何其他操作（如保存到日志）。这样，响应应该更快。

使用 gevent，您可以实现并行运行处理程序的不同实例。因此，如果您有单个请求，您不会看到响应时间有任何差异，但如果您有数千个并发请求，则性能会好得多。这样做的原因是，没有 gevent，当你的代码执行 IO 操作时，它会阻止该进程的执行，而使用 gevent，CPU 可以在 IO 操作等待时继续执行其他请求。

设置 gevent 比设置 Celery 容易得多。如果您使用的是 gunicorn，您只需安装 gevent 并将工作器类型更改为 gevent。另一个优点是您可以并行化响应中所需的任何操作（例如从数据库中提取响应）。在 Celery 中，您不能在响应中使用 Celery 任务的输出。

我建议先使用 gevent，然后考虑在以下情况下添加 Celery（并同时使用它们）：

  • 响应中不需要您将使用 Celery 处理的任务的输出
  • 您的 celery 任务使用不同的机器，或者您的服务器的使用有一些高峰和一些空闲时间（如果您的服务器一直处于 100% 的状态，那么使用 Celery 不会有任何好处）
  • 您的 Celery 任务将完成的工作量值得使用 Celery 的开销
于 2015-12-21T16:23:02.957 回答
3

您可以使用multiprocessing.Processwithdeamon=True运行守护进程并立即向调用者返回响应：

from multiprocessing import Process

class APIHandler:

  def on_get(self, req, resp):
    heavy_process = Process(  # Create a daemonic process
        target=my_func,
        daemon=True
    )
    heavy_process.start()
    resp.body = "Quick response"


# Define some heavy function
def my_func():
    time.sleep(10)
    print("Process finished")

您可以通过发送GET请求来测试它。您将立即收到回复，10 秒后您将在控制台中看到一条打印消息。

于 2020-03-27T15:41:33.250 回答