私は次のシステムを持っています: [Client] - [Web Server] - [Connector]。
コネクタは、Web サーバーとデータ ソースの間の一種の中間コードです。
コネクタとのサーバー接続を監視する必要があります。接続が失われた場合は、クライアントに通知する必要があります。
Web サーバーとコネクタ間の通信は、socketio を使用して編成されます。
問題は、コネクタが機能しなくなった場合、Web サーバーがそのことを 1 分後にしか認識できないことです (これが最善のケースです)。
私は、サーバーがコネクタのステータスを毎秒チェックする必要があると判断しました。
コネクタがサーバーに接続されると、バックグラウンド タスクが開始されます。タスクの本質: 毎秒: 1) 時間を修正します。2) 固定時間をスタックに保存します。3) コネクタにエコー メッセージを送信します。(server.background_thread を参照)
コネクタは、エコー メッセージとタイム スタンプをパラメーターとして受け取り、受信したタイム スタンプをパラメーターとして渡すことで、エコー メッセージを Web サーバーに送信します。(client.echo を参照)
タイムスタンプがスタック内の最後の値と等しい場合、Web サーバーはエコー メッセージを受信し、この値はスタックから削除されます。(server.on_echo_connector を参照)
Web サーバーでは、反復ごとにスタック サイズがチェックされます (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()
パイソン 2.7