python - 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 tries to (ab)use the Copy On Write semantics of the fork() call whenever possible. By default it will fork after having loaded your applications to share as much of their memory as possible. If this behavior is undesirable for some reason, use the lazy option. This will instruct uWSGI to load the applications after each worker’s fork(). Lazy mode changes the way graceful reloading works: instead of reloading the whole instance, each worker is reloaded in chain. If you want “lazy app loading”, but want to maintain the standard uWSGI reloading behaviour, starting from 1.3 you can use the lazy-apps option.

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

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