如何在不依赖其他软件的情况下为多个工作人员的 Flask 应用程序提供共享状态？

Question

我想为与多个工作人员（即多个进程）一起运行的 Flask 应用程序提供共享状态。

从关于这个主题的类似问题中引用这个答案：

您不能使用全局变量来保存此类数据。[...] 使用 Flask 之外的数据源来保存全局数据。根据您的需要，数据库、memcached 或 redis 都是合适的独立存储区域。

（来源：flask 中的全局变量线程安全吗？如何在请求之间共享数据？）

我的问题是关于如何在 Flask 的“外部”提供数据的建议的最后一部分。目前，我的网络应用程序非常小，我想避免对其他程序的要求或依赖。如果我不想在后台运行 Redis 或其他任何东西，而是提供 Web 应用程序的 Python 代码的所有内容，我有哪些选择？

Answer 1

如果您的网络服务器的工作类型与multiprocessing模块兼容，您可以使用multiprocessing.managers.BaseManager为 Python 对象提供共享状态。一个简单的包装器可能如下所示：

from multiprocessing import Lock
from multiprocessing.managers import AcquirerProxy, BaseManager, DictProxy

def get_shared_state(host, port, key):
    shared_dict = {}
    shared_lock = Lock()
    manager = BaseManager((host, port), key)
    manager.register("get_dict", lambda: shared_dict, DictProxy)
    manager.register("get_lock", lambda: shared_lock, AcquirerProxy)
    try:
        manager.get_server()
        manager.start()
    except OSError:  # Address already in use
        manager.connect()
    return manager.get_dict(), manager.get_lock()

您可以将数据分配给shared_dict以使其跨进程访问：

HOST = "127.0.0.1"
PORT = 35791
KEY = b"secret"
shared_dict, shared_lock = get_shared_state(HOST, PORT, KEY)

shared_dict["number"] = 0
shared_dict["text"] = "Hello World"
shared_dict["array"] = numpy.array([1, 2, 3])

但是，您应该注意以下情况：

• 用于shared_lock在覆盖shared_dict. （请参阅下面的烧瓶示例。）
• 没有数据持久性。如果您重新启动应用程序，或者如果主（第一个）BaseManager进程终止，则共享状态将消失。
• 使用 的这种简单实现BaseManager，您无法直接编辑 中的嵌套值shared_dict。例如，shared_dict["array"][1] = 0没有效果。您必须编辑一个副本，然后将其重新分配给字典键。

烧瓶示例：

以下 Flask 应用程序使用全局变量来存储计数器编号：

from flask import Flask
app = Flask(__name__)

number = 0

@app.route("/")
def counter():
    global number
    number += 1
    return str(number)

这在仅使用 1 个 worker 时有效gunicorn -w 1 server:app。当使用多个 workergunicorn -w 4 server:app时，很明显这number不是共享状态，而是每个 worker 进程的单独状态。

相反，使用shared_dict，应用程序看起来像这样：

from flask import Flask
app = Flask(__name__)

HOST = "127.0.0.1"
PORT = 35791
KEY = b"secret"
shared_dict, shared_lock = get_shared_state(HOST, PORT, KEY)

shared_dict["number"] = 0

@app.route("/")
def counter():
    with shared_lock:
        shared_dict["number"] += 1
    return str(shared_dict["number"])

这适用于任意数量的工人，例如gunicorn -w 4 server:app.