Flask：后台线程将非空队列视为空

Question

当我在 uwsgi 中运行 Flask 应用程序时，后台线程和应用程序函数在查询同一队列的大小时会看到不同的值。

组件

• 带有thread-safe queue 的 Flask 应用程序。
• GET 调用返回队列大小。
• POST 调用将一个元素添加到队列中。
• 后台线程打印队列大小

问题

当应用程序来自使用 python tester.py 的 shell 时，我得到了预期的结果：

2014-06-07 14:20:50.677995 Queue size is: 0
127.0.0.1 - - [07/Jun/2014 14:20:51] "POST /addMessage/X HTTP/1.1" 200 -
2014-06-07 14:20:51.679277 Queue size is: 1
2014-06-07 14:20:52.680425 Queue size is: 1
2014-06-07 14:20:53.681566 Queue size is: 1
2014-06-07 14:20:54.682708 Queue size is: 1
127.0.0.1 - - [07/Jun/2014 14:20:55] "POST /addMessage/Y HTTP/1.1" 200 -
2014-06-07 14:20:55.687755 Queue size is: 2
2014-06-07 14:20:56.688867 Queue size is: 2

但是，当应用程序使用 uwsgi 执行时，我在日志中得到以下信息：

2014-06-07 14:17:42.056863 Queue size is: 0
2014-06-07 14:17:43.057952 Queue size is: 0
[pid: 9879|app: 0|req: 6/6] 127.0.0.1 () {24 vars in 280 bytes} [Sat Jun  7 14:17:43 2014] POST /addMessage/X => generated 16 bytes in 0 msecs (HTTP/1.1 200) 2 headers in 71 bytes (1 switches on core 0)
2014-06-07 14:17:44.059037 Queue size is: 0
2014-06-07 14:17:45.060118 Queue size is: 0
[pid: 9879|app: 0|req: 7/7] 127.0.0.1 () {24 vars in 280 bytes} [Sat Jun  7 14:17:45 2014] POST /addMessage/X => generated 16 bytes in 0 msecs (HTTP/1.1 200) 2 headers in 71 bytes (1 switches on core 0)
2014-06-07 14:17:46.061205 Queue size is: 0
2014-06-07 14:17:47.062286 Queue size is: 0

在uwsgi下运行时，后台线程看不到与app相同的队列。这是为什么？如何让这两个线程查看同一个 Queue 对象？

更新

• 即使作为 Python 脚本执行，我也看到不一致的行为：有时它无法记录消息（使用 app.logger），我只能看到 prints。这意味着线程正在运行，但它无法对app.logger 执行任何操作。

uwsgi.ini配置

[uwsgi]
http-socket    = :9002
plugin         = python
wsgi-file      = /home/ubuntu/threadtest-uwsgi.py
enable-threads = true
workers        = 1
chdir          = /home/ubuntu/thread-tester/thread_tester

代码

from flask import Flask, jsonify
import Queue
from threading import Thread
import time
import datetime
import logging
import sys

logging.basicConfig(stream=sys.stderr,
                    format='%(asctime)s %(levelname)s - %(message)s')

app = Flask(__name__)
messages = Queue.Queue()

def print_queue_size():
    while True:
        app.logger.debug("%s Queue size is: %d" % (datetime.datetime.now(),
                                        messages.qsize()))
        time.sleep(1)

t = Thread(target=print_queue_size, args=())
t.setDaemon(True)
t.start()

@app.route("/queueSize", methods=["GET"])
def get_queue_size():
    return jsonify({"qsize": messages.qsize()}), 200

@app.route("/addMessage/<message>", methods=["POST"])
def add_message_to_queue(message):
    messages.put(message)
    return jsonify({"qsize": messages.qsize()}), 200

if __name__ == "__main__":
    app.run(port=6000)

Answer 1

来自Things to Know documenation page：

uWSGI 尽可能尝试（ab）使用 fork() 调用的 Copy On Write 语义。默认情况下，它会在加载您的应用程序以共享尽可能多的内存后分叉。如果由于某种原因不希望出现这种行为，请使用lazy 选项。这将指示 uWSGI 在每个 worker 的 fork() 之后加载应用程序。惰性模式改变了优雅重新加载的工作方式：不是重新加载整个实例，而是链式重新加载每个工作人员。如果您想要“延迟应用加载”，但想要保持标准的 uWSGI 重新加载行为，从 1.3 开始，您可以使用 lazy-apps 选项。

你的 Flask 应用程序在 uWSGI 启动时启动，然后一个工作进程被分叉。分叉时，Queue 对象为空，不再与原始进程共享。线程没有被带走。

尝试设置 lazy-apps option 以延迟 Flask 应用的加载，直到工作程序启动。

Answer 2

@Martijn Pieters 的回答中的文档链接指出，lazy-apps 可能比预分叉消耗更多的内存。如果您对此感到担忧，您可能还希望考虑使用 @postfork 装饰器来更精细地控制分叉后运行的内容。您可以在 @postfork-decorated 函数中创建队列，它将在每个工作人员中创建。