Python: flask-socketio使用Websocket协议进行通讯

文档：

PyPI: https://pypi.org/project/Flask-SocketIO/

Github: https://github.com/miguelgrinberg/Flask-SocketIO

doc: https://flask-socketio.readthedocs.io

socket.io: https://socket.io/

安装

pip install flask-socketio gevent-websocket

1

代码实例

from flask import Flask, render_template, request from flask_socketio import SocketIO app = Flask(__name__) app.config['SECRET_KEY'] = 'secret!' app.jinja_env.auto_reload = True socketio = SocketIO(app) @app.route('/') def index(): return render_template('index.html') @app.route('/send_all') def send_all(): """ 广播 :return: """ message = request.args.get('message') socketio.send(message) return {'status': 'ok'} @app.route('/send') def send_message(): """ 单独发送 :return: """ sid = request.args.get('sid') message = request.args.get('message') socketio.send(message, to=sid) return {'status': 'ok'} @socketio.on('connect') def connect(): print('connect') socketio.send({'sid': request.sid}) @socketio.on('disconnect') def disconnect(): print('disconnect') @socketio.on('message') def handle_message(data): print(data) socketio.send({'data': data}) if __name__ == '__main__': socketio.run(app, debug=True)

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

Http 发送测试请求

import requests params = { 'sid': 'm4AymrH2TIFHCcQNAAAF', 'message': 'send' } res = requests.get('http://localhost:5000/send', params=params) print(res.text) params = { 'message': 'send_all' } res = requests.get('http://localhost:5000/send_all', params=params) print(res.text)

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

Flask Python TCP/IP

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。