1. Toy模板网首页
  2. > Toy博客

Websocket 之flask_socketio和Python socketio

7月前 作者：爱补鱼的猫猫 分类：Toy博客 阅读(30) 违法举报

这篇具有很好参考价值的文章主要介绍了Websocket 之flask_socketio和Python socketio。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

1、falsk_socket与前端

（1）测试版

前端

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Title</title>


   <!--
    <script type="text/javascript" src="https://cdn.bootcdn.net/ajax/libs/socket.io/2.3.0/socket.io.js"></script>
-->

  <script src="https://cdnjs.cloudflare.com/ajax/libs/jquery/3.5.1/jquery.min.js" integrity="sha512-bLT0Qm9VnAYZDflyKcBaQ2gg0hSYNQrJ8RilYldYQ1FxQYoCLtUjuuRuZo+fjqhx/qtq/1itJ0C2ejDxltZVFg==" crossorigin="anonymous"></script>

    <script src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/3.0.4/socket.io.js" integrity="sha512-aMGMvNYu8Ue4G+fHa359jcPb1u+ytAF+P2SCb+PxrjCdO3n3ZTxJ30zuH39rimUggmTwmh2u7wvQsDTHESnmfQ==" crossorigin="anonymous"></script>



</head>
<body>

 <div><button onclick="ws()">连接服务端</button></div>
 <div><button onclick="clos_con()">断开连接</button></div>
 <div><button onclick="send_msg()">发送消息给服务端</button></div>
<script type="text/javascript">
    windw_socket=null


    function ws(){
	namespace = '/websocket';
	var websocket_url = location.protocol+'//' + document.domain + ':' + location.port + namespace;
	var socket=io.connect(websocket_url);
	// socket.emit('connect2', {'param':'value'});	//发送消息
	// socket.close()
	socket.on('connect',function(data){
		console.log('connecte:'+data);
		alert("建立连接成功")
		windw_socket=socket
	});

	socket.on('disconnect',function(data){
		alert("连接已断开")
		console.log('disconnecte:'+data);
	});

	socket.on('my_response_message',function(data){
		console.log('my_response_message:'+data);
		alert("收到服务端的回复:"+data)
	});
}



    function clos_con(){
	    if(windw_socket!=null){
	        windw_socket.close()
        }
    }



    window.onbeforeunload= function(event) {
	if (windw_socket!=null && !windw_socket.closed){
		// confirm(windw_socket.closed)
		windw_socket.close()
	}
    }


    window.onunload= function(event) {
	if (windw_socket!=null && !windw_socket.closed){
		//confirm(windw_socket.closed)
		windw_socket.close()
	}
    }


    function send_msg(){
	    if(windw_socket!=null){
	        windw_socket.emit('message', "这里是客户端");
        }
    }


</script>
</body>
</html>

后端

import json
from flask import Flask, render_template, request, redirect, sessions
from flask_socketio import SocketIO, emit  # 导入socketio包

name_space = '/websocket'
app = Flask(__name__)
app.secret_key = 'jzw'
socketio = SocketIO(app)
client_query = []


@socketio.on('connect', namespace=name_space)  # 有客户端连接会触发该函数
def on_connect():
    # 建立连接 sid:连接对象ID
    client_id = request.sid
    client_query.append(client_id)
    # emit(event_name, broadcasted_data, broadcast=False, namespace=name_space, room=client_id)  #指定一个客户端发送消息
    # emit(event_name, broadcasted_data, broadcast=True, namespace=name_space)  #对name_space下的所有客户端发送消息
    print('有新连接,id=%s接加入, 当前连接数%d' % (client_id, len(client_query)))


@socketio.on('disconnect', namespace=name_space)  # 有客户端断开WebSocket会触发该函数
def on_disconnect():
    # 连接对象关闭 删除对象ID
    client_query.remove(request.sid)
    print('有连接,id=%s接退出, 当前连接数%d' % (request.sid, len(client_query)))


# on('消息订阅对象', '命名空间区分')
@socketio.on('message', namespace=name_space)
def on_message(message):
    """ 服务端接收消息 """
    print('从id=%s客户端中收到消息，内容如下:' % request.sid)
    print(message)
    emit('my_response_message', "我收到了你的信息", broadcast=False, namespace=name_space, room=client_id)  # 指定一个客户端发送消息
    # emit('my_response_message', broadcasted_data, broadcast=True, namespace=name_space)  #对name_space下的所有客户端发送消息


@app.route('/')  # 初始化页面
def a():
    return render_template("socketv5.html")


if __name__ == '__main__':
    socketio.run(app, host='0.0.0.0', port=5000, debug=False)
    # app.run()

（2）完整版

前端

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>SocketIO Demo</title>

<!--
    <script type="text/javascript" src="//cdn.bootcss.com/jquery/3.1.1/jquery.min.js"></script>
    <script type="text/javascript" src="//cdn.bootcss.com/socket.io/1.5.1/socket.io.min.js"></script>
-->

    <!--
     <script src="https://cdnjs.cloudflare.com/ajax/libs/jquery/3.5.1/jquery.min.js" integrity="sha512-bLT0Qm9VnAYZDflyKcBaQ2gg0hSYNQrJ8RilYldYQ1FxQYoCLtUjuuRuZo+fjqhx/qtq/1itJ0C2ejDxltZVFg==" crossorigin="anonymous"></script>

    <script src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/3.0.4/socket.io.js" integrity="sha512-aMGMvNYu8Ue4G+fHa359jcPb1u+ytAF+P2SCb+PxrjCdO3n3ZTxJ30zuH39rimUggmTwmh2u7wvQsDTHESnmfQ==" crossorigin="anonymous"></script>

<script src="https://cdn.staticfile.org/jquery/1.10.2/jquery.min.js"></script>

-->

     <script src="https://cdnjs.cloudflare.com/ajax/libs/jquery/3.5.1/jquery.min.js" integrity="sha512-bLT0Qm9VnAYZDflyKcBaQ2gg0hSYNQrJ8RilYldYQ1FxQYoCLtUjuuRuZo+fjqhx/qtq/1itJ0C2ejDxltZVFg==" crossorigin="anonymous"></script>

    <script src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/3.0.4/socket.io.js" integrity="sha512-aMGMvNYu8Ue4G+fHa359jcPb1u+ytAF+P2SCb+PxrjCdO3n3ZTxJ30zuH39rimUggmTwmh2u7wvQsDTHESnmfQ==" crossorigin="anonymous"></script>



</head>
<body>

<h2>Demo of SocketIO</h2>
<div id="t"></div>

<button onclick="myconet()">建立连接</button> <br>
<button onclick="send()">发送信息</button> <br>
<button onclick="getTest()">点ajax请求</button> <br>
<button onclick="getTest2()">点server服务端关闭连接</button> <br>
<button onclick="clos_con()">点client客户端关闭连接</button> <br>




<script>

    namespace = '/dcenter';
    var socket=null;

    {#var socket = io.connect(location.protocol + '//' + document.domain + ':' + location.port + namespace);#}

    socket = io.connect(location.protocol + '//' + document.domain + ':' + location.port + namespace);


    socket.on('dcenter', function (res) {
        var t = res.data;
         console.log("dcenter test:");
         console.log(t);
        if (t) {
            $("#t").append(t).append('<br/>');
        }

    });

    socket.on('my_response', function (res) {
        var t = res.data;
         console.log(t);
        if (t) {
            $("#t").append(t).append('<br/>');
        }

    });

    socket.on('disconnect',function(data){
		alert("连接已断开")
		console.log('disconnecte:'+data);
	});

    function myconet() {
        {#socket=null;#}
        socket = io.connect(location.protocol + '//' + document.domain + ':' + location.port + namespace);

     {#location.reload(true);#}
    window.location.reload(true) // 刷新窗口
    console.log('my connecte:');

    };


    function send(){
            socket.emit('my_event', { 'data': 'I\'m a buttun ok !' });
             console.log('button click')
        };

    function getTest(){
        socket.emit('my_event', { 'data': 'I\'m a ajax go go !' });
             console.log('button click')

	};

	 function getTest2(){
        socket.emit('leav', { 'data': 'I\'m a leav !' });
             console.log('button click')

	};

	function getTest3(){
        socket.emit('disconnect', { 'data': 'I\'m a leav !' });
             console.log('disconnect click')

	};

	function clos_con(){
	    if(socket!=null){
	        socket.close()
        }
    }


</script>


</body>
</html>

后端

from flask import Flask, render_template,request
from flask_socketio import SocketIO, emit


app = Flask(__name__)
app.config['SECRET_KEY'] = 'secret_key'

socketio = SocketIO()

socketio.init_app(app, cors_allowed_origins='*')

name_space = '/dcenter'


@app.route('/')
def index():
    return render_template('socketv4.html')


@app.route('/push')
def push_once():
    event_name = 'dcenter'
    broadcasted_data = {'data': "test message!"}
    socketio.emit(event_name, broadcasted_data, broadcast=False, namespace=name_space)
    return 'done!'


@socketio.on('connect', namespace=name_space)
def connected_msg():
    print('client connected.')


@socketio.on('disconnect', namespace=name_space)
def disconnect_msg():
    print('client disconnected.')


@socketio.on('my_event', namespace=name_space)
def mtest_message(message):
    print(message)

    # socketio.emit('my_response',
    #      {'data': message['data'], 'count': 1},broadcast=False, namespace=name_space)

    socketio.emit('my_response',{'data': 'i am server you kow'},broadcast=False, namespace=name_space)


@app.route('/push2')
def push_once2():
    event_name = 'dcenter'
    broadcasted_data = {'data': "test message!"}
    socketio.emit(event_name, broadcasted_data, broadcast=False, namespace=name_space)
    return 'done!'



@socketio.on('leav', namespace=name_space)
def receive_message(msg):
    print('i leav pre')
    socketio.close_room(None,name_space)
    print('i leav you')


# @socketio.on('disconnect', namespace=name_space)  # 有客户端断开WebSocket会触发该函数
# def on_disconnect():
#     # 连接对象关闭 删除对象ID
#
#     print('有连接,id=%s接退出当前连接' % (request.sid))





if __name__ == '__main__':

    socketio.run(app,host='127.0.0.1',port=5000, debug=True) #端口在配置文件中配置 --port=8080,默认为5000

2、client与serve（socket）

（1）python的socketio实现

client

import socketio

sio = socketio.Client()

@sio.event
def connect():
    print('connection established')


@sio.event
def my_message(data):
    print('message received with ', data)
    # sio.emit('my response', {'response': 'my response'})

@sio.event
def disconnect():
    print('disconnected from server')

sio.connect('http://127.0.0.1:5000')

# sio.connect('http://127.0.0.1:5000', namespaces=['/chat'])

print("000")

sio.emit('my_message', {'response': 'my response'})

sio.sleep(10)

sio.emit('my_message', {'response': 'my response22222222'})

# sio.emit('myname', {'foo': 'a name space test'}, namespace='/chat')

print("222")

sio.wait()

server

import eventlet
import socketio

sio = socketio.Server()
app = socketio.WSGIApp(sio, static_files={
    '/': {'content_type': 'text/html', 'filename': 'index.html'}
})

@sio.event
def connect(sid, environ):
    print('connect ', sid)


@sio.event
def disconnect(sid):
    print('disconnect ', sid)

@sio.event
def my_message(sid, data):    #没名称空间用方法名
    print('message ', data)
    sio.emit('my_message', {'response': 'my response'})
    # sio.emit('my message', {'response': 'my response'})
    # sio.emit('my event', {'data': 'foobar'})

# @sio.on('myname', namespace='/chat')
# def my_custom_event(sid, data):
#     print('message ', data)

if __name__ == '__main__':
    eventlet.wsgi.server(eventlet.listen(('127.0.0.1', 5000)), app)

（2）flask和socketio客户端实现

client

import socketio

sio = socketio.Client()

@sio.event
def connect():
    print('connection established')


@sio.event
def disconnect():
    print('disconnected from server')



@sio.event
def my_message(data):
    print('message received with ', data)
    # sio.emit('my response', {'response': 'my response'})


@sio.event(namespace='/dcenter')
def my_response(data):
    print('message received with ', data)
    # sio.emit('my response', {'response': 'my response'})



@sio.on('test_event', namespace='/dcenter')
def test_event(data):
    print(data)
    print("I'm connected to the /chat namespace!")


# sio.connect('http://127.0.0.1:5000')

sio.connect('http://127.0.0.1:5000', namespaces=['/dcenter'])


print("000")
sio.emit('my_event', {'foo': 'a name space test'}, namespace='/dcenter')

sio.sleep(2)
sio.emit('my_event', {'foo': 'a name space2222'}, namespace='/dcenter')

print("222")
sio.sleep(2)
sio.emit('test_event', {'event': 'a serve test event'}, namespace='/dcenter')

print('3333')
# print("000")
#
# sio.emit('my_message', {'response': 'my response'})
#
# sio.sleep(10)
#
# sio.emit('my_message', {'response': 'my response22222222'})
#
# # sio.emit('myname', {'foo': 'a name space test'}, namespace='/chat')
#
# print("222")

sio.wait()

Flask server

from flask import Flask, render_template,request
from flask_socketio import SocketIO, emit


app = Flask(__name__)
app.config['SECRET_KEY'] = 'secret_key'

socketio = SocketIO()

socketio.init_app(app, cors_allowed_origins='*')

name_space = '/dcenter'


@app.route('/')
def index():
    return render_template('socketv4.html')


@app.route('/push')
def push_once():
    event_name = 'dcenter'
    broadcasted_data = {'data': "test message!"}
    socketio.emit(event_name, broadcasted_data, broadcast=False, namespace=name_space)
    return 'done!'


@socketio.on('connect', namespace=name_space)
def connected_msg():
    print('client connected.')


@socketio.on('disconnect', namespace=name_space)
def disconnect_msg():
    print('client disconnected.')


@socketio.on('my_event', namespace=name_space)
def mtest_message(message):
    print(message)

    # socketio.emit('my_response',
    #      {'data': message['data'], 'count': 1},broadcast=False, namespace=name_space)

    socketio.emit('my_response',{'data': 'i am server you kow'},broadcast=False, namespace=name_space)


@socketio.on('test_event', namespace=name_space)
def test_event(message):
    print(message)

    # socketio.emit('my_response',
    #      {'data': message['data'], 'count': 1},broadcast=False, namespace=name_space)

    socketio.emit('test_event',{'data': 'i am client test_event'},broadcast=False, namespace=name_space)



@app.route('/push2')
def push_once2():
    event_name = 'dcenter'
    broadcasted_data = {'data': "test message!"}
    socketio.emit(event_name, broadcasted_data, broadcast=False, namespace=name_space)
    return 'done!'



@socketio.on('leav', namespace=name_space)
def receive_message(msg):
    print('i leav pre')
    socketio.close_room(None,name_space)
    print('i leav you')


# @socketio.on('disconnect', namespace=name_space)  # 有客户端断开WebSocket会触发该函数
# def on_disconnect():
#     # 连接对象关闭 删除对象ID
#
#     print('有连接,id=%s接退出当前连接' % (request.sid))

if __name__ == '__main__':

    socketio.run(app,host='127.0.0.1',port=5000, debug=True) #端口在配置文件中配置 --port=8080,默认为5000

3、Python环境

bidict==0.22.1
certifi==2023.5.7
cffi==1.15.1
charset-normalizer==3.1.0
click==8.1.3
colorama==0.4.6
dnspython==2.3.0
eventlet==0.33.3
Flask==2.2.2
Flask-SocketIO==5.2.0
gevent==22.10.2
greenlet==2.0.2
h11==0.14.0
idna==3.4
itsdangerous==2.1.2
Jinja2==3.1.2
MarkupSafe==2.1.1
pip==23.1.2
pycparser==2.21
python-engineio==4.3.3
python-socketio==5.7.1
requests==2.31.0
setuptools==67.8.0
simple-websocket==0.7.0
six==1.16.0
urllib3==2.0.3
Werkzeug==2.2.2
wheel==0.38.4
wsproto==1.2.0
zope.event==4.6
zope.interface==6.0

版本问题参考：https://blog.csdn.net/weixin_44986037/article/details/131504760文章来源地址https://www.toymoban.com/news/detail-529182.html

到了这里，关于Websocket 之flask_socketio和Python socketio的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处： 如若内容造成侵权/违法违规/事实不符，请点击违法举报进行投诉反馈，一经查实，立即删除！

分享到：
领支付宝红包 赞助服务器费用

相关文章

  • Python使用flask框架与前端建立websocket链接，并进行数据交互

    Python使用flask框架与前端建立websocket链接，并进行数据交互

    后端采用的框架为flask，前端用的flask自带的html编写，实现的功能为：前后端建立websocket链接并进行数据交互 一、编写一个flask后端服务 常规创建方式就可以，创建一个flask服务。声明一个websocket实例，以websocket的方式启动这个服务。 安装相关依赖： 声明应用和websocket实例

    2024年04月11日
    浏览(35)

  • Flask-SocketIO和Flask-Login联合开发socketio权限系统

    设置 Flask, Flask-SocketIO, Flask-Login : 首先，确保安装了必要的库: 基础设置 : 定义登录路由 : 定义 SocketIO 路由并进行身份验证 : 简单的登录页面 ( templates/login.html ) : 简单的首页 ( templates/index.html ) : 创建一个简单的首页 ( index.html )，用户登录后可以发送消息到服务器，然后接收服

    2024年02月11日
    浏览(28)
  • Flask SocketIO 实现动态绘图

    Flask SocketIO 实现动态绘图

    Flask-SocketIO 是基于 Flask 的一个扩展，用于简化在 Flask 应用中集成 WebSocket 功能。WebSocket 是一种在客户端和服务器之间实现实时双向通信的协议，常用于实现实时性要求较高的应用，如聊天应用、实时通知等，使得开发者可以更轻松地构建实时性要求较高的应用。通过定义事

    2024年02月05日
    浏览(30)
  • Flask 学习99-Flask-SocketIO 快速入门与使用

    Flask 学习99-Flask-SocketIO 快速入门与使用

    flask-socketio 为flask应用提供了一个客户端与服务器之间低延迟的双向通讯 官网地址：https://flask-socketio.readthedocs.io/en/latest/intro.html 先安装flask-socketio 说明Flask-SocketIO 与 js版本客户端不匹配，二者不能正常通信。兼容版本说明：https://flask-socketio.readthedocs.io/en/latest/intro.html 我安装

    2024年02月22日
    浏览(38)

  • Flask 学习100-Flask-SocketIO 结合 xterm.js 实现网页版Xshell

    xterm.js 是一个使用 TypeScript 编写的前端终端组件，可以直接在浏览器中实现一个命令行终端应用。 可以实现 web-terminal 功能，类似于Xshell 操作服务器。 Flask-SocketIO 快速入门与使用基础参考前面这篇https://www.cnblogs.com/yoyoketang/p/18022139 前端代码 io.connect 连上服务端 socket.emit 给后

    2024年02月21日
    浏览(34)

  • flask socketio 实时传值至html上【需补充实例】

    跨域问题网上的普通方法无法解决。 参考这篇文章解决 Flask教程(十九)SocketIO - 迷途小书童的Note迷途小书童的Note (xugaoxiang.com) 如果用了eventlet，在while true中，emit数据不在前端web页面显示，可参考这篇文章的解决方案，加定时器。 javascript - Flask-websocket Emit主动发送消息失败

    2024年02月07日
    浏览(32)
  • 【快速开始】一个简单的Flask-SocketIO应用，完成后端推送消息接收与关闭

    【快速开始】一个简单的Flask-SocketIO应用，完成后端推送消息接收与关闭

    本人使用环境及版本： Anaconda： 虚拟环境： Python版本：3.8.13 安装包及版本： Flask-SocketIO ：5.3.4 eventlet ：0.33.3 创建app.py文件（文件名随意，不过要与后面的运行脚本中指定的文件保持一致） cmd 或者 linux控制台运行即可 此时能看到如下图所示 此时访问http://0.0.0.0:5200（0.0.0

    2024年02月13日
    浏览(41)

  • 【ChatGPT】如何使用Qt设计SocketIO类型的WebSocket协议客户端

    本文第一次发布 2023年03月24日09点28分 Authors ChatGPT / THDMI 最近写 Flask-SocketIO 服务端的时候，苦于不熟悉前端代码，想试着用 QT 来写客户端，但不清楚该怎么描述，以致于好像一直查的都是 RAW 类型的 WebSocket 协议处理，于是一直没找到 event 和 room 或者 namespace 处理方式。 简单

    2023年04月21日
    浏览(52)
  • 进度变动实时通知-使用SocketIO实现前后端的通信（基于WebSocket的实时通信库）

    进度变动实时通知-使用SocketIO实现前后端的通信（基于WebSocket的实时通信库）

    最近在接触的一个项目，将PDF上传到项目里，通过调用OCR云服务把PDF里的表格数据识别出来。在此过程中，前后端需要实时通信，对识别数据进行“进度跟踪”。因此我们采用SocketIO的通讯方式，识别中前端和后端服务建立SocketIO连接，根据事件进行数据的实时更新百分比进度

    2024年02月06日
    浏览(40)

  • Python Flask flasgger api文档[python/flask/flasgger]

    首先需要安装依赖： 封装swagger.py文件，代码如下：  然后在主应用中（项目入口文件）加入以下代码： 然后在user.py文件中加入以下代码： 运行项目， 访问：http://127.0.0.1:5000/apidocs/

    2024年02月10日
    浏览(25)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包，优惠每天领

二维码1

领取红包

二维码2

领红包