所有 Web 客户端都未收到服务器发送的事件

Question

我有一个 Flask 网络服务器，它生成服务器发送事件 (sse)，所有连接的网络客户端都应该接收这些事件。

在下面的“版本 1”中有效。所有网络客户端都会收到事件，并相应地更新。

在下面的“第 2 版”中，这是对第 1 版的重构，这不再按预期工作：

相反，我得到：

• 大多数情况下只有一个 Web 客户端会收到该事件，或者
• 很少 多个网络客户端获取事件，或
• 很少有 网络客户端收到事件

据我所知，服务器始终在生成事件，并且通常至少有一个客户端正在接收。

我的初始测试将网络服务器托管在 Raspberry Pi 3 上，网络客户端在 Pi 上，在 Windows 和 OSX 上使用各种浏览器。

为了消除任何可能的网络问题，我对 Web 服务器和 3 个 Chrome 实例重复了相同的测试，这些实例都托管在同一台 OSX 笔记本电脑上。 这给出了相同的结果：版本 1“OK”，版本 2“NOT OK”。

成功接收的客户端似乎因事件而异：到目前为止，我无法辨别模式。

版本 1 和版本 2 都有一个结构 change_objects，其中包含“应跟踪更改的内容”

• 在版本 1 中，change_objects 是一个字典。

• 在版本 2 中，我将 change_objects 重构为 Reporter 类或 Reporter 的子类的实例列表。

对“事物”的更改是根据代码中其他地方收到的网络服务触发的。

版本 1（OK：所有网络客户端都收到 sse 事件）

def check_walk(walk_new, walk_old):
    if walk_new != walk_old:
        print("walk change", walk_old, walk_new)
        return True, walk_new
    else:
        return False, walk_old

def walk_event(walk):
    silliness = walk['silliness']
    data = '{{"type": "walk_change", "silliness": {}}}'.format(silliness)
    return "data: {}\n\n".format(data)

change_objects = {
    "walk1": {
        "object": walks[0],
        "checker": check_walk,
        "event": walk_event,
    },
    ... more things to be tracked...
}

def event_stream(change_objects):
    copies = {}
    for key, value in change_objects.items():
        copies[key] = {"obj_old": deepcopy(value["object"])}  # ensure a true copy, not a reference!

    while True:
        gevent.sleep(0.5)
        for key, value in change_objects.items():
            obj_new = deepcopy(value["object"]) # use same version in check and yield functions
            obj_changed, copies[key]["obj_old"] = value["checker"](obj_new, copies[key]["obj_old"])
            if (obj_changed):
                yield value["event"](obj_new)

@app.route('/server_events')
def sse_request():
    return Response(
            event_stream(change_objects),
            mimetype='text/event-stream')

第 2 版（不正常：并非所有网络客户端都收到 sse 事件）

class Reporter:

    def __init__(self, reportee, name):
        self._setup(reportee, name)

    def _setup(self, reportee, name):
        self.old = self.truecopy(reportee)
        self.new = reportee
        self.name = "{}_change".format(name)

    def truecopy(self, orig):
        return deepcopy(orig)

    def changed(self):
        if self.new != self.old:
            self.old = self.truecopy(self.new)
            return True
        else:
            return False

    def sse_event(self):
        data = self.new.copy()
        data['type'] = self.name
        data = json.dumps(data)
        return "data: {}\n\n".format(data)

class WalkReporter(Reporter):

    # as we are only interested in changes to attribute "silliness" (not other attributes) --> override superclass sse_event
    def sse_event(self): 
        silliness = self.new['silliness']
        data = '{{"type": "walk_change", "silliness": {}}}'.format(silliness)
        return "data: {}\n\n".format(data)

change_objects = [
    WalkReporter(name="walk1", reportee=walks[0]),
    ... more objects to be tracked...
] 

def event_stream(change_objects):
    while True:
        gevent.sleep(0.5)
        for obj in change_objects:
            if obj.changed():
                yield obj.sse_event()

@app.route('/server_events')
def sse_request():
    return Response(
            event_stream(change_objects),
            mimetype='text/event-stream')

完全披露：这个问题是问题的后续：Refactor a (multi)generator python function 它专注于在跟踪多个“事物”的更改时重构 event_stream() 函数。 但是这里的问题显然超出了原始问题的范围，因此是一个新问题。

Answer 1

问题中重构的“版本 2”代码存在并发/计时问题。

sse_request() 为每个 Web 客户端调用（在测试用例 3 实例中）。因此，我们有 3 个实例在 event_stream() 中循环。

这些调用“或多或少”并行发生：这实际上意味着随机顺序。

但是列表 change_objects 是共享的，因此第一个发现更改的 Web 客户端会将共享的 WalkReporter 实例中的“旧”副本更新为最新状态，并且可能会在其他客户端发现之前这样做改变。即第一个成功的网络客户端有效地隐藏了其他网络客户端的更改。

这很容易解决，方法是为每个 Web 客户端提供自己的 change_objects 副本。

即change_objects 移动到sse_request() 中，如下所示。

@app.route('/server_events')
def sse_request():
    change_objects = [
        WalkReporter(name="walk1", reportee=walks[0]),
        ... more objects to be tracked...
    ]
    return Response(
            event_stream(change_objects),
            mimetype='text/event-stream')

通过这个微小的更改，sse_request() 的每个实例都可以发现更改，因此所有 Web 客户端都可以按预期接收 sse 事件。