为什么 flask_socketio.disconnect 会阻止线程完成?
Why does flask_socketio.disconnect prevent the thread from completing?
我有以下系统:[Client] - [Web Server] - [Connecotr].
连接器是网络服务器和数据源之间的一种中间代码。
我需要用连接器监视服务器连接。如果连接丢失,那么我必须通知客户端。
Web 服务器和连接器之间的通信是使用 socketio 组织的。
问题是如果连接器停止工作,Web 服务器将在一分钟后才知道它(这是最好的情况)。
我决定服务器应该每秒检查一次连接器的状态。
连接器连接到服务器后,后台任务启动。任务本质:每一秒:1)固定时间; 2)将固定时间存入栈; 3) 向连接器发送回显消息。 (参见 server.background_thread)
连接器接受回显消息和时间戳作为参数,并向 Web 服务器发送回显消息,作为参数传递接收到的时间戳。 (参见 client.echo)
网络服务器收到回显消息,如果时间戳等于堆栈中的最后一个值,则从堆栈中删除该值。 (参见 server.on_echo_connector)
在网络服务器上,每次迭代都会检查堆栈大小(参见 server.background_thread)。如果大于5,则表示连接器5次没有响应回显信息,我们认为连接器不可用,断开连接。
当服务器意识到连接器不可用时,有必要终止向连接器发送回显消息的线程。
一旦堆栈大小大于 5,我就退出无限循环并调用 flask_socketio.disconnect (connector_sid, '/ connector')
。在这个调用之后没有任何效果(例如 print
)
在 on_disconnect_connector
(服务器)方法中,thread.join()
被调用并且永不终止。
而且我需要完成线程,这样当连接器再次启动时,它会连接成功,一切从头再来。
如何解决这个问题?
服务器
# -*- coding: utf-8 -*-
import os
import threading
import time
import collections
from datetime import datetime
import flask
import flask_socketio
def get_unix_time():
return int(time.mktime(datetime.now().timetuple()))
class Stack(collections.deque):
def __init__(self, iterable=(), maxlen=None):
collections.deque.__init__(self, iterable, maxlen)
@property
def size(self):
return len(self)
@property
def empty(self):
return self.size == 0
@property
def head(self):
return self[-1]
@property
def tail(self):
return self[0]
def push(self, x):
self.append(x)
# SERVER
app = flask.Flask(__name__)
sio = flask_socketio.SocketIO(app, async_mode='gevent')
connector_sid = None
echo_stack = Stack()
thread = None
thread_lock = threading.Lock()
def background_thread(app):
time.sleep(2) # delay for normal connection
while True:
if echo_stack.size >= 5:
break
time_ = get_unix_time()
echo_stack.push(time_)
sio.emit('echo', time_, namespace='/connector')
sio.sleep(1)
with app.app_context():
flask_socketio.disconnect(connector_sid, '/connector')
@sio.on('connect', namespace='/connector')
def on_connect_connector():
"""Connector connection event handler."""
global connector_sid, thread
print 'Attempt to connect a connector {}...'.format(request.sid)
# if the connector is already connected, reject the connection
if connector_sid is not None:
print 'Connection for connector {} rejected'.format(request.sid)
return False
# raise flask_socketio.ConnectionRefusedError('Connector already connected')
connector_sid = request.sid
print('Connector {} connected'.format(request.sid))
with thread_lock:
if thread is None:
thread = sio.start_background_task(
background_thread, current_app._get_current_object())
# notify clients about connecting a connector
sio.emit('set_connector_status', True, namespace='/client')
@sio.on('disconnect', namespace='/connector')
def on_disconnect_connector():
"""Connector disconnect event handler."""
global connector_sid, thread
print 'start join'
thread.join()
print 'end join'
thread = None
print 'after disconet:', thread
connector_sid = None
echo_stack.clear()
print('Connector {} disconnect'.format(request.sid))
# notify clients of disconnected connector
sio.emit('set_connector_status', False, namespace='/client')
@sio.on('echo', namespace='/connector')
def on_echo_connector(time_):
if not echo_stack.empty:
if echo_stack.head == time_:
echo_stack.pop()
@sio.on('message', namespace='/connector')
def on_message_connector(cnt):
# print 'Msg: {}'.format(cnt)
pass
if __name__ == '__main__':
sio.run(app)
客户
# -*- coding: utf-8 -*-
import sys
import threading
import time
import socketio
import socketio.exceptions
sio = socketio.Client()
thread = None
thread_lock = threading.Lock()
work = False
def background_thread():
# example task
cnt = 0
while work:
cnt += 1
if cnt % 10 == 0:
sio.emit('message', cnt // 10, namespace='/connector')
sio.sleep(0.1)
@sio.on('connect', namespace='/connector')
def on_connect():
"""Server connection event handler."""
global thread, work
print '\n----- Connected to server -----' \
'\n----- My SID: {} -----\n'.format(sio.sid)
work = True # set flag
# run test task
with thread_lock:
if thread is None:
thread = sio.start_background_task(background_thread)
@sio.on('disconnect', namespace='/connector')
def on_disconnect():
"""Server disconnect event handler."""
global thread, work
# clear the work flag so that at the next iteration the endless loop ends
work = False
thread.join()
thread = None
# disconnect from server
sio.disconnect()
print '\n----- Disconnected from server -----\n'
# switch to the mode of infinite attempts to connect to the server
main()
@sio.on('echo', namespace='/connector')
def on_echo(time_):
sio.emit('echo', time_, namespace='/connector')
def main():
while True:
try:
sio.connect('http://localhost:5000/connector',
namespaces=['/connector'])
sio.wait()
except socketio.exceptions.ConnectionError:
print 'Trying to connect to the server...'
time.sleep(1)
except KeyboardInterrupt:
print '\n---------- EXIT ---------\n'
sys.exit()
except Exception as e:
print e
if __name__ == '__main__':
print '\n---------- START CLIENT ----------\n'
main()
Python 2.7
需要为客户端安装额外的库(see)
pip install "python-socketio[client]"
多亏了这个库,WebSocket 传输才能正常工作。现在断开连接器是立即可见的。
我有以下系统:[Client] - [Web Server] - [Connecotr].
连接器是网络服务器和数据源之间的一种中间代码。
我需要用连接器监视服务器连接。如果连接丢失,那么我必须通知客户端。
Web 服务器和连接器之间的通信是使用 socketio 组织的。
问题是如果连接器停止工作,Web 服务器将在一分钟后才知道它(这是最好的情况)。
我决定服务器应该每秒检查一次连接器的状态。
连接器连接到服务器后,后台任务启动。任务本质:每一秒:1)固定时间; 2)将固定时间存入栈; 3) 向连接器发送回显消息。 (参见 server.background_thread)
连接器接受回显消息和时间戳作为参数,并向 Web 服务器发送回显消息,作为参数传递接收到的时间戳。 (参见 client.echo)
网络服务器收到回显消息,如果时间戳等于堆栈中的最后一个值,则从堆栈中删除该值。 (参见 server.on_echo_connector)
在网络服务器上,每次迭代都会检查堆栈大小(参见 server.background_thread)。如果大于5,则表示连接器5次没有响应回显信息,我们认为连接器不可用,断开连接。
当服务器意识到连接器不可用时,有必要终止向连接器发送回显消息的线程。
一旦堆栈大小大于 5,我就退出无限循环并调用 flask_socketio.disconnect (connector_sid, '/ connector')
。在这个调用之后没有任何效果(例如 print
)
在 on_disconnect_connector
(服务器)方法中,thread.join()
被调用并且永不终止。
而且我需要完成线程,这样当连接器再次启动时,它会连接成功,一切从头再来。
如何解决这个问题?
服务器
# -*- coding: utf-8 -*-
import os
import threading
import time
import collections
from datetime import datetime
import flask
import flask_socketio
def get_unix_time():
return int(time.mktime(datetime.now().timetuple()))
class Stack(collections.deque):
def __init__(self, iterable=(), maxlen=None):
collections.deque.__init__(self, iterable, maxlen)
@property
def size(self):
return len(self)
@property
def empty(self):
return self.size == 0
@property
def head(self):
return self[-1]
@property
def tail(self):
return self[0]
def push(self, x):
self.append(x)
# SERVER
app = flask.Flask(__name__)
sio = flask_socketio.SocketIO(app, async_mode='gevent')
connector_sid = None
echo_stack = Stack()
thread = None
thread_lock = threading.Lock()
def background_thread(app):
time.sleep(2) # delay for normal connection
while True:
if echo_stack.size >= 5:
break
time_ = get_unix_time()
echo_stack.push(time_)
sio.emit('echo', time_, namespace='/connector')
sio.sleep(1)
with app.app_context():
flask_socketio.disconnect(connector_sid, '/connector')
@sio.on('connect', namespace='/connector')
def on_connect_connector():
"""Connector connection event handler."""
global connector_sid, thread
print 'Attempt to connect a connector {}...'.format(request.sid)
# if the connector is already connected, reject the connection
if connector_sid is not None:
print 'Connection for connector {} rejected'.format(request.sid)
return False
# raise flask_socketio.ConnectionRefusedError('Connector already connected')
connector_sid = request.sid
print('Connector {} connected'.format(request.sid))
with thread_lock:
if thread is None:
thread = sio.start_background_task(
background_thread, current_app._get_current_object())
# notify clients about connecting a connector
sio.emit('set_connector_status', True, namespace='/client')
@sio.on('disconnect', namespace='/connector')
def on_disconnect_connector():
"""Connector disconnect event handler."""
global connector_sid, thread
print 'start join'
thread.join()
print 'end join'
thread = None
print 'after disconet:', thread
connector_sid = None
echo_stack.clear()
print('Connector {} disconnect'.format(request.sid))
# notify clients of disconnected connector
sio.emit('set_connector_status', False, namespace='/client')
@sio.on('echo', namespace='/connector')
def on_echo_connector(time_):
if not echo_stack.empty:
if echo_stack.head == time_:
echo_stack.pop()
@sio.on('message', namespace='/connector')
def on_message_connector(cnt):
# print 'Msg: {}'.format(cnt)
pass
if __name__ == '__main__':
sio.run(app)
客户
# -*- coding: utf-8 -*-
import sys
import threading
import time
import socketio
import socketio.exceptions
sio = socketio.Client()
thread = None
thread_lock = threading.Lock()
work = False
def background_thread():
# example task
cnt = 0
while work:
cnt += 1
if cnt % 10 == 0:
sio.emit('message', cnt // 10, namespace='/connector')
sio.sleep(0.1)
@sio.on('connect', namespace='/connector')
def on_connect():
"""Server connection event handler."""
global thread, work
print '\n----- Connected to server -----' \
'\n----- My SID: {} -----\n'.format(sio.sid)
work = True # set flag
# run test task
with thread_lock:
if thread is None:
thread = sio.start_background_task(background_thread)
@sio.on('disconnect', namespace='/connector')
def on_disconnect():
"""Server disconnect event handler."""
global thread, work
# clear the work flag so that at the next iteration the endless loop ends
work = False
thread.join()
thread = None
# disconnect from server
sio.disconnect()
print '\n----- Disconnected from server -----\n'
# switch to the mode of infinite attempts to connect to the server
main()
@sio.on('echo', namespace='/connector')
def on_echo(time_):
sio.emit('echo', time_, namespace='/connector')
def main():
while True:
try:
sio.connect('http://localhost:5000/connector',
namespaces=['/connector'])
sio.wait()
except socketio.exceptions.ConnectionError:
print 'Trying to connect to the server...'
time.sleep(1)
except KeyboardInterrupt:
print '\n---------- EXIT ---------\n'
sys.exit()
except Exception as e:
print e
if __name__ == '__main__':
print '\n---------- START CLIENT ----------\n'
main()
Python 2.7
需要为客户端安装额外的库(see)
pip install "python-socketio[client]"
多亏了这个库,WebSocket 传输才能正常工作。现在断开连接器是立即可见的。