Python Flask服务器，如何向另一个请求的线程发送消息？

Question

我有一个使用Python编写的Flask Web服务器，它会在接收每个请求时执行一些耗费资源的任务。例如：

1) CLIENT --> REQUEST1 --> SERVER
2) Server performing expensive task for request1...
3) CLIENT --> REQUEST2 --> SERVER
4) Server Performing expensive tasks for request1 and request2 in parallel...
5) Server sends complete response for request1 (or request2)
6) Server sends complete response for request2 (or request1)

我希望如果有第二个请求到达，服务器能够停止执行第一个请求（request1），以下是我所需的服务器行为：

1) CLIENT --> REQUEST1 --> SERVER
2) Server performing expensive tasks for request1...
3) CLIENT --> REQUEST2 --> SERVER
4) Server immediately sends (incomplete) response for request1
5) Server performing expensive tasks for request2...
6) Server sends complete response for request2 

如何在request2 worker中获取对request1 worker的引用，以便向其发送停止消息？ 在flask中是否可以实现这一点（或者必要时利用一些实用程序库）？

- Giuseppe Galano

---

你使用的是哪个版本的Python？如果是2.x，你尝试过gevent吗？ - Anton

我正在使用Python 3.x，并尝试在Mac OS X Mavericks上使用pip3安装gevent，但未成功。 - Giuseppe Galano

嗯...你看过这个页面了吗？链接 - Anton

Answer 1

这不是一个答案，但我无法在评论中清楚地解释我的想法。

有一个不太美观的解决方案。 假设您有一段代码，在请求到达服务器时执行。

def funcReq()

但是，你可以创建一个新的线程，然后执行 funcReq()，这样你就有一个变量来暂停你感兴趣的线程。你可以将它存储在某个地方（我相信 Flask 有一个全局对象，它会传递给每个请求），以便所有请求都能访问到它。现在，你就有了控制执行 Request 1 的线程的能力，从 Request 2 中。

这种解决方法的问题是你最终需要创建和运行两倍数量的线程。而且这很丑陋。

- Anton

---

我想让Flask（在线程模式下）来管理线程。我想要获取由Flask创建而不是我创建的线程的引用。可能这是不可实现的，我需要在单线程中使用Flask和像gevent这样的库来管理线程。 - Giuseppe Galano

我最终采用了你的解决方案，因此我接受你的答案！ - Giuseppe Galano

Answer 2

我成功地使用了Andrey建议的队列和线程库中的事件来实现我想要的功能。 该事件称为“worker_stopped”，如果没有请求的工作程序，则会设置该事件。 队列称为“worker_to_worker”，用于从新工作程序（Request2的工作程序）向旧工作程序（request1的工作程序）发送“停止”消息。

同步是通过每个请求开头的以下代码完成的。

if (not worker_stopped.isSet()):
  print('ordering the current worker to stop')
  worker_to_worker.put('stop')
  print('waiting that current worker has stopped')
  worker_stopped.wait()
worker_stopped.clear()
perform_expensive_task()

并定期检查 worker_to_worker 队列中是否有消息

if (worker_to_worker.empty()):
    # continue...
else:
    print('I\'m consuming a message from the worker_to_worker queue')
    worker_to_worker.get()
    worker_to_worker.task_done()
    print('signaling that the worker stopped')
    worker_stopped.set()
    # break...

- Giuseppe Galano