前言:

websocket和socketIO是全然不同的兩個東西,websocket的話 使用自帶ws的庫就可以完成一些功能,但是socketIO屬于sw的另外一塊內(nèi)容

工作中遇到了一個監(jiān)控socketIO傳輸?shù)牧奶煨畔⒈O(jiān)控的需求.研究了一陣 話不多說 上代碼!

from socketIO_client import SocketIO, BaseNamespace

import time

import requests

import logging

from app.test_report.constant import VIEWER_DATA, TEACHER_DATA, VIEWER_DATA_MUTIL, TEACHER_DATA_MUTIL # 這里引入四個變量 放在flask里面起一個單獨文件防止有坑!

logging.getLogger('socketIO-client').setLevel(logging.DEBUG)

logging.basicConfig() # 調(diào)試時候可以自定義日志

class TeacherNamespace(BaseNamespace):

def on_teacher_response(self, *args):

print('qqq', args, type(args))

class ViewerNamespace(BaseNamespace):

def on_viewer_response(self, *args):

print(args, type(args))

def get_sessionid(userid, roomid):

pass

def on_connect(*args):

# print(*args)

print('connect')

def on_disconnect():

print('## disconnect ##')

def on_reconnect():

print('reconnect')

def revc_message(*args): # 這個函數(shù)很重要 是socketio監(jiān)聽信息獲取的函數(shù), 這里面socketio會自行調(diào)用這個函數(shù)

response_data_format = {}

response_data = eval(args[0])

print(response_data)

def revc_message_mutil(*args): # 這個函數(shù)很重要 是socketio監(jiān)聽信息獲取的函數(shù), 這里面socketio會自行調(diào)用這個函數(shù)

response_data_format = {}

response_data = eval(args[0])

print(response_data)

def create_params(userid,roomid): # 這個方法類似requests庫的封裝方法 看底層貌似與requests的urllib3方法一樣

vsessionid, tsessionid = get_sessionid(userid, roomid)

vparams = {

'sessionid': vsessionid,

}

tparams = {

'sessionid': tsessionid,

}

return vparams, tparams

def sio(userid, roomid, host=' xxx.net', vparams='', tparams=''):

if not vparams and not tparams:

vparams, tparams = create_params(userid, roomid)

vsocketIO = SocketIO(host, params=vparams)

tsocketIO = SocketIO(host, params=tparams)

viewer = vsocketIO.define(ViewerNamespace, path=f'/asjbfasbfk') # 這個path類似于信道的路徑 很重要

teacher = tsocketIO.define(TeacherNamespace, path=f'/asjbfasbfk') # 這個path類似于信道的路徑 很重要

viewer.on('connect', on_connect)

viewer.on('chat_message', revc_message)

viewer.on('disconnect', on_disconnect)

viewer.on('reconnect', on_reconnect)

teacher.on('connect', on_connect)

teacher.on('chat_message', revc_message)

teacher.on('disconnect', on_disconnect)

teacher.on('reconnect', on_reconnect)

return teacher, vsocketIO, tsocketIO

def send_msg(domain_host, featurestype, sendmessage, userid, roomid, timer=1):

teacher, vsocketIO, tsocketIO = sio(userid, roomid, host=domain_host)

for i in range(timer):

teacher.emit('chat_message', f'{sendmessage}', revc_message)

vsocketIO.wait(seconds=1)

tsocketIO.wait(seconds=1)

return VIEWER_DATA, TEACHER_DATA

def send_msg_mutil(domain_host, featurestype, sendmessage, userid, roomid, timer=1, vparams='', tparams=''):

teacher, vsocketIO, tsocketIO = sio(userid, roomid, host=domain_host, vparams=vparams, tparams=tparams)

for i in range(timer):

teacher.emit('chat_message', f'{sendmessage}', revc_message_mutil)

vsocketIO.wait(seconds=1)

tsocketIO.wait(seconds=1)

return VIEWER_DATA, TEACHER_DATA

def clear_data():

VIEWER_DATA.clear()

TEACHER_DATA.clear()

if __name__ == '__main__':

s = time.time()

print(send_msg('xxxx.net', 1, '發(fā)送信息1', '7848DB3F76g7ts7057F', 'E301DFFFCDDAG7S79GY9A01307461'))

print(time.time()-s)

clear_data()

總結(jié)

以上是生活随笔為你收集整理的python socketio_python3--socketIO_client 摸索怕坑指南的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。