Flask-SocketIO服務端

安裝Flask-Sockets很容易：

pip install flask-socketio

注意Flask-SocketIO依賴gevent庫，目前它僅可以在python2上運行（譯者注：python3.6測試也可以）。gevent很快也會對python3支持。

下面是一個Flask-SocketIO在Flask應用上的實現例子：

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


app = Flask(__name__)
app.config['SECRET_KEY'] = 'secret!'
socketio = SocketIO(app)


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

#觸發事件my event：回復只發送此連接
@socketio.on('my_event', namespace='/chat')
def test_message(message):
    print(message)
    emit('my_response', {'data': message['data']})


#觸發事件my broadcast event：：回復所有鏈接（廣播）
@socketio.on('my_broadcast event', namespace='/chat')
def test_message(message):
    print(message)
    emit('my_response', {'data': message['data']}, broadcast=True)

##################################################################
#自動連接和自動斷開觸發
@socketio.on('connect', namespace='/chat')
def test_connect():
    print("連接到來")
    emit('my_response', {'data': 'Connected'})

@socketio.on('disconnect', namespace='/chat')
def test_disconnect():
    print("連接斷開")
    print('Client disconnected')

if __name__ == '__main__':
    socketio.run(app,host='0.0.0.0',port=8200,debug=True)

對應的test.html,注意一個問題：flask是從與啟動文件同級的templates查找模板的。

<!DOCTYPE HTML>
<html>
<head>
    <title>Flask-SocketIO Test</title>
    <script src="https://cdn.bootcdn.net/ajax/libs/jquery/1.12.4/jquery.min.js"></script>
    <script src="//cdnjs.cloudflare.com/ajax/libs/socket.io/2.2.0/socket.io.js" integrity="sha256-yr4fRk/GU1ehYJPAs8P4JlTgu0Hdsp4ZKrx8bDEDC3I=" crossorigin="anonymous"></script>
    <script type="text/javascript" charset="utf-8">
            $(document).ready(function(){
         var socket = io.connect('http://' + document.domain + ':' + location.port + '/chat');
             socket.on('my_response', function(msg) {
                 $('#log').append('<p>Received: ' + msg.data + '</p>');
        });

            $('form#emit').submit(function(event) {
                socket.emit('my_event', {data: $('#emit_data').val()});
                return false;
            });

            $('form#broadcast').submit(function(event) {
                socket.emit('my_broadcast event', {data:$('#broadcast_data').val()});
                return false;
            });

    });
    </script>
</head>
<body>
    <h1>Flask-SocketIO Test</h1>
    <p>Async mode is: <b>{{ async_mode }}</b></p>
    <p>Average ping/pong latency: <b><span id="ping-pong"></span>ms</b></p>
    <h2>Send:</h2>
    <form id="emit" method="POST" action='#'>
        <input type="text" name="emit_data" id="emit_data" placeholder="Message">
        <input type="submit" value="Echo">
    </form>
    <form id="broadcast" method="POST" action='#'>
        <input type="text" name="broadcast_data" id="broadcast_data" placeholder="Message">
        <input type="submit" value="Broadcast">
    </form>

    <form id="disconnect" method="POST" action="#">
        <input type="submit" value="Disconnect">
    </form>
    <h2>Receive:</h2>
    <div id="log"></div>
</body>
</html>

test.html

官方完整示例

#!/usr/bin/env python
from threading import Lock
from flask import Flask, render_template, session, request, \
    copy_current_request_context
from flask_socketio import SocketIO, emit, join_room, leave_room, \
    close_room, rooms, disconnect

# Set this variable to "threading", "eventlet" or "gevent" to test the
# different async modes, or leave it set to None for the application to choose
# the best option based on installed packages.
async_mode = None

app = Flask(__name__)
app.config['SECRET_KEY'] = 'secret!'
socketio = SocketIO(app, async_mode=async_mode)
thread = None
thread_lock = Lock()

def background_thread():
    """Example of how to send server generated events to clients."""
    count = 0
    while True:
        socketio.sleep(100)
        count += 1
        socketio.emit('my_response',
                      {'data': 'Server generated event', 'count': count},
                      namespace='/test')

@app.route('/')
def index():
    return render_template('test1.html', async_mode=socketio.async_mode)

@socketio.on('my_event', namespace='/test')
def mtest_message(message):
    print(message)
    session['receive_count'] = session.get('receive_count', 0) + 1
    # print(message)
    # print(message['data'])
    emit('my_response',
         {'data': message['data'], 'count': session['receive_count']})

@socketio.on('my_broadcast_event', namespace='/test')
def mtest_broadcast_message(message):
    print(message)
    session['receive_count'] = session.get('receive_count', 0) + 1
    emit('my_response',
         {'data': message['data'], 'count': session['receive_count']},
         broadcast=True)

@socketio.on('join', namespace='/test')
def join(message):
    print(message)
    join_room(message['room'])
    session['receive_count'] = session.get('receive_count', 0) + 1
    emit('my_response',
         {'data': 'In rooms: ' + ', '.join(rooms()),
          'count': session['receive_count']})

@socketio.on('leave', namespace='/test')
def leave(message):
    print(message)
    leave_room(message['room'])
    session['receive_count'] = session.get('receive_count', 0) + 1
    emit('my_response',
         {'data': 'In rooms: ' + ', '.join(rooms()),
          'count': session['receive_count']})

@socketio.on('close_room', namespace='/test')
def close(message):
    session['receive_count'] = session.get('receive_count', 0) + 1
    emit('my_response', {'data': 'Room ' + message['room'] + ' is closing.',
                         'count': session['receive_count']},
         room=message['room'])
    close_room(message['room'])

@socketio.on('my_room_event', namespace='/test')
def send_room_message(message):
    session['receive_count'] = session.get('receive_count', 0) + 1
    emit('my_response',
         {'data': message['data'], 'count': session['receive_count']},
         room=message['room'])


##################################################################
#自動連接和自動斷開觸發
@socketio.on('connect', namespace='/test')
def test_connect():
    print("連接到來")
    emit('my response', {'data': 'Connected'})

@socketio.on('disconnect', namespace='/test')
def test_disconnect():
    print("連接斷開")
    print('Client disconnected')


if __name__ == '__main__':
    socketio.run(app, debug=True,port= 8200)

<!DOCTYPE HTML>
<html>
<head>
    <title>Flask-SocketIO Test</title>
    <script src="https://cdn.bootcdn.net/ajax/libs/jquery/1.12.4/jquery.min.js"></script>
    <script src="//cdnjs.cloudflare.com/ajax/libs/socket.io/2.2.0/socket.io.js" integrity="sha256-yr4fRk/GU1ehYJPAs8P4JlTgu0Hdsp4ZKrx8bDEDC3I=" crossorigin="anonymous"></script>
    <script type="text/javascript" charset="utf-8">
            $(document).ready(function(){
         var socket = io.connect('http://' + document.domain + ':' + location.port + '/test');
             socket.on('my_response', function(msg) {
                 $('#log').append('<p>Received: ' + msg.data + '</p>');
        });

            $('form#emit').submit(function(event) {
                socket.emit('my_event', {data: $('#emit_data').val()});
                return false;
            });
            $('form#broadcast').submit(function(event) {
                socket.emit('my_broadcast_event', {data: $('#broadcast_data').val()});
                return false;
            });
            $('form#join').submit(function(event) {
                socket.emit('join', {room: $('#join_room').val()});
                return false;
            });
            $('form#leave').submit(function(event) {
                socket.emit('leave', {room: $('#leave_room').val()});
                return false;
            });
            $('form#send_room').submit(function(event) {
                socket.emit('my_room_event', {room: $('#room_name').val(), data: $('#room_data').val()});
                return false;
            });
            $('form#close').submit(function(event) {
                socket.emit('close_room', {room: $('#close_room').val()});
                return false;
            });
            $('form#disconnect').submit(function(event) {
                socket.emit('disconnect_request');
                return false;
            });
        });
    </script>
</head>
<body>
    <h1>Flask-SocketIO Test</h1>
    <p>Async mode is: <b>{{ async_mode }}</b></p>
    <p>Average ping/pong latency: <b><span id="ping-pong"></span>ms</b></p>
    <h2>Send:</h2>
    <form id="emit" method="POST" action='#'>
        <input type="text" name="emit_data" id="emit_data" placeholder="Message">
        <input type="submit" value="Echo">
    </form>
    <form id="broadcast" method="POST" action='#'>
        <input type="text" name="broadcast_data" id="broadcast_data" placeholder="Message">
        <input type="submit" value="Broadcast">
    </form>
    <form id="join" method="POST" action='#'>
        <input type="text" name="join_room" id="join_room" placeholder="Room Name">
        <input type="submit" value="Join Room">
    </form>
    <form id="leave" method="POST" action='#'>
        <input type="text" name="leave_room" id="leave_room" placeholder="Room Name">
        <input type="submit" value="Leave Room">
    </form>
    <form id="send_room" method="POST" action='#'>
        <input type="text" name="room_name" id="room_name" placeholder="Room Name">
        <input type="text" name="room_data" id="room_data" placeholder="Message">
        <input type="submit" value="Send to Room">
    </form>
    <form id="close" method="POST" action="#">
        <input type="text" name="close_room" id="close_room" placeholder="Room Name">
        <input type="submit" value="Close Room">
    </form>
    <form id="disconnect" method="POST" action="#">
        <input type="submit" value="Disconnect">
    </form>
    <h2>Receive:</h2>
    <div id="log"></div>
</body>
</html>

test1.html

免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。

猜您在找 Flask中使用websocket Python Flask使用自帶的websocket WebSocket with Flask “Easy WebSockets with Flask and Gevent” 翻譯學習筆記 ——Flask WebSocket使用快速入門 flask二十五: websocket WebSocket 的使用 WebSocket的使用 flask基本使用 Flask基於websocket的簡單聊天室 flask與Flask-CORS的使用