gunicorn 与 gevent 工作人员：使用共享的全局列表答案

【问题标题】：gunicorn with gevent workers: Using a shared global listgunicorn 与 gevent 工作人员：使用共享的全局列表
【发布时间】：2015-05-10 12:09:21
【问题描述】：

我正在尝试按照以下简单方法在我的 Flask 应用程序中实现服务器发送事件：http://flask.pocoo.org/snippets/116/

为了为应用程序提供服务，我将 gunicorn 与 gevent 工作人员一起使用。

我的代码的最小版本如下所示：

import multiprocessing

from gevent.queue import Queue
from gunicorn.app.base import BaseApplication
from flask import Flask, Response

app = Flask('minimal')
# NOTE: This is the global list of subscribers
subscriptions = []


class ServerSentEvent(object):
    def __init__(self, data):
        self.data = data
        self.event = None
        self.id = None
        self.desc_map = {
            self.data: "data",
            self.event: "event",
            self.id: "id"
        }

    def encode(self):
        if not self.data:
            return ""
        lines = ["%s: %s" % (v, k)
                 for k, v in self.desc_map.iteritems() if k]
        return "%s\n\n" % "\n".join(lines)


@app.route('/api/events')
def subscribe_events():
    def gen():
        q = Queue()
        print "New subscription!"
        subscriptions.append(q)
        print len(subscriptions)
        print id(subscriptions)
        try:
            while True:
                print "Waiting for data"
                result = q.get()
                print "Got data: " + result
                ev = ServerSentEvent(unicode(result))
                yield ev.encode()
        except GeneratorExit:
            print "Removing subscription"
            subscriptions.remove(q)
    return Response(gen(), mimetype="text/event-stream")


@app.route('/api/test')
def push_event():
    print len(subscriptions)
    print id(subscriptions)
    for sub in subscriptions:
        sub.put("test")
    return "OK"


class GunicornApplication(BaseApplication):
    def __init__(self, wsgi_app, port=5000):
        self.options = {
            'bind': "0.0.0.0:{port}".format(port=port),
            'workers': multiprocessing.cpu_count() + 1,
            'worker_class': 'gevent',
            'preload_app': True,
        }
        self.application = wsgi_app
        super(GunicornApplication, self).__init__()

    def load_config(self):
        config = dict([(key, value) for key, value in self.options.iteritems()
                       if key in self.cfg.settings and value is not None])
        for key, value in config.iteritems():
            self.cfg.set(key.lower(), value)

    def load(self):
        return self.application


if __name__ == '__main__':
    gapp = GunicornApplication(app)
    gapp.run()

问题在于订阅者的列表似乎对每个工人都不同。这意味着如果 worker #1 处理 /api/events 端点并将新订阅者添加到列表中，客户端将仅接收在 worker #1 也处理 /api/test 端点时添加的事件。

奇怪的是，每个工作人员的实际列表对象似乎都是相同的，因为id(subscriptions) 在每个工作人员中返回相同的值。

有没有办法解决这个问题？我知道我可以只使用 Redis，但应用程序应该尽可能自包含，所以我试图避免任何外部服务。

更新： ~~问题的原因似乎在于我嵌入了gunicorn.app.base.BaseApplication（即new feature in v0.19）。从命令行使用gunicorn -k gevent minimal:app 运行应用程序时，一切正常~~

更新 2： 之前的怀疑被证明是错误的，它起作用的唯一原因是因为 gunicorn 的默认工作进程数是 1，当通过-w 参数，它表现出相同的行为。

【问题讨论】：

标签： python gunicorn gevent

【解决方案1】：

你说：

每个工人的实际列表对象似乎都是相同的，因为 id(subscriptions) 在每个 worker 中返回相同的值。

但我认为这不是真的，每个工人上的subscriptions 不是同一个对象。每个worker都是一个独立的进程，有自己的内存空间。

对于自包含系统，您可以开发一个小型系统，其功能类似于Redis 的简单版本。例如，使用 SQLite 或 ZeroMQ 在这些工作人员之间进行通信。

【讨论】：