Flask socketio多线程转储字典以排队并发送给客户端

Question

目标仅仅是能够使字典的线程队列并将其报告给客户端。

编辑这与Flask throwing 'working outside of request context' when starting sub thread不同，因为：

不是在路由功能中完成，而是在socketio.start_background_task中完成

唯一的socketio代码在上下文中发生，使用socketio.emit我们正在发送字典。

策略：在服务器端有2个不同的任务，每个任务都要构建一个线程，然后在另一个socketio线程中收集结果，这些结果存储在词典的线程安全队列FIFO中。

然后将这些词典发送给客户端，并等待每个确认。

因此现在减少问题来解决：RuntimeError: Working outside of request context.

from flask import Flask, flash, request, redirect, render_template, Response, escape, jsonify, url_for, session, copy_current_request_context
#socketio
from flask_socketio import SocketIO, send, emit, join_room, leave_room, close_room, rooms, disconnect
import threading
from threading import Thread, Event, Lock
import queue
import random

def ack(value):
    if value != 'pong':
        logger.info('unexpected return value')

def fn_i():
    global q
    while True:
        time.sleep(1)
        q.put({'key_i':random.random()})
        return q

def fn_ii():
    global q
    while True:
        time.sleep(10)
        q.put({'key_ii':random.random()})
        return q

app = Flask(__name__)
socketio = SocketIO(app, async_mode=async_mode)
thread1=None
thread2=None
collector_thread=None
q = queue.Queue()
thread_lock = Lock()

def background_thread_collector():
    global thread1
    global thread2
    global q

    thread1 = threading.Thread(target=fn_i)
    thread1.start() 

    thread2 = threading.Thread(target=fn_ii)
    thread2.start() 

    """Example of how to send server generated events to clients."""
    while True:
        time.sleep(0.2)
        while not q.empty():
            socketio.emit('my_response',
                          q.get(), #{'data': 'Server generated event', 'count': count},
                          namespace='/test',
                          broadcast=True,
                          callback=ack
                         )

@app.route('/')
def index():
    return render_template('index.html', async_mode=socketio.async_mode)

@socketio.on('connect', namespace='/test')
def test_connect():
    global collector_thread
    logger.info(' Client connected ' + request.sid)
    with thread_lock:
        if collector_thread is None:
            collector_thread = socketio.start_background_task(background_thread_collector)            
    emit('my_response', {'data': 'Connected', 'count': 0})
if __name__ == '__main__':
    socketio.run(app, 
                host='localhost',
                 port=10000, 
                 debug=False) #True sends some exceptions and stops)

欢呼声

Answer 1

这应该由Flask-SocketIO更好地处理，但是问题是您试图在设置为广播给所有客户端的发射上使用回调：

            socketio.emit('my_response',
                          q.get(), #{'data': 'Server generated event', 'count': count},
                          namespace='/test',
                          broadcast=True,
                          callback=ack
                         )

删除回调，发射应该可以正常工作。

Answer 2

与Flasksocketio一起使用线程队列并不简单，因为需要处理应用程序上下文，为了在客户端刷新服务器日志文件，发现在这种情况下简单地使用javascript更容易，并且可以相应地更新文件。