python心跳测试代码_flask 如何实现一个心跳检测接口

使用flask提供一个心跳检测接口

当客户端访问该接口时，服务端就会隔一段时间发送心跳包进行检测客户端是否在线。

我使用的是flask-socketio模块进行检测，可以连接，我在本地运行，打开两个浏览器器时，另外一个就会掉线，只留了一个，我也用了Thread模块，但是使用Thread模块不知道客户端下线了之后，如何停止那个线程。还有需要多个客户端进行连接是该怎么做.

下面是我参照网上写的,但是感觉客户端多了之后会出问题，然后也不知道该怎么去做性能测试

后台代码:

class MyCustomNamespace(Namespace):

clients = []

def on_connect(self):

sid = request.sid

print('client connected websocket: {}'.format(sid))

self.clients.append(sid)

global thread

with thread_lock:

if thread is None:

thread = socketio.start_background_task(target=self.background_thread)

def on_disconnect(self):

sid = request.sid

print('close websocket: {}'.format(sid))

self.clients.remove(sid)

def on_message(self, data):

print('received message: ' + data["param"])

if data["param"] is not None:

print('websocket {} is alive'.format(request.sid))

# else:

# print('websocket {} is die'.format(*rooms()))

# self.disconnect(*rooms())

def background_thread(self):

while True:

time.sleep(6)

data = {"status": "ok"}

socketio.emit('server_detect', data, namespace='/testnamespace', broadcast=True)

print("服务器探测客户端是否还活着")

socketio.on_namespace(MyCustomNamespace("/testnamespace"))

前端代码:

$(document).ready(function () {

namespace = '/testnamespace';

webSocketUrl = location.protocol + '//' + document.domain + ':' + location.port + namespace;

console.log(webSocketUrl);

var socket = io.connect(webSocketUrl);

socket.on('server_detect', function (res) {

console.log(res);

socket.emit('message',{'param':'value'});

});

});