你想了解的轮询、长轮询和websocket都在这里了

提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

日常生活中，有很多需要数据的实时更新，比如群聊信息的实时更新，还有投票系统的实时刷新等

实现的方式有很多种，比如轮询、长轮询、websocket

轮询

轮询是通过设置页面的刷新频率（设置多长时间自动刷新一次页面）来实现的。

使用轮询的机制模拟投票系统的实时刷新

<!DOCTYPE html>

<html lang="en">

<head>

    <meta charset="UTF-8">

    <meta name="viewport" content="width=device-width, initial-scale=1">

    <title>Title</title>

</head>

<body>

    <ul>

        {% for k,v in dic.items() %}

            <li style="cursor:pointer" ondblclick="ticket('{{k}}')">{{v.name}}:{{v.count}}</li>

        {% endfor %}

    </ul>

</body>

<script src="/static/jquery-3.3.1.js"></script>

<script>

     function ticket(nid){

        $.ajax({

            url:'/ticket',

            type:'post',

            data:{nid:nid},

            success:function (arg) {

            }

        })

    }

    setInterval(function () {

        window.location.reload();

    },2000);

</script>

</html>

HTML

备注：可以给标签添加样式：style="cursor:pointer"，点击时，鼠标样式为小手。

from flask import Flask,request,render_template

app = Flask(__name__,static_folder='static',template_folder='template')

app.debug = True

dic = {'':{'name':'a','count':0},'':{'name':'b','count':0}}

@app.route('/index',methods=['GET','POST'])

def index():

    return render_template('index.html',dic=dic)

@app.route('/ticket',methods=['POST'])

def ticket():

    if request.method == 'POST':

        nid = request.form.get('nid')

        print(nid)

        dic[nid]['count'] += 1

    return '投票成功'

if __name__ == '__main__':

    app.run(host='127.0.0.1',port=5000)

.py

缺点：需要频繁的发送请求，服务端压力大，如果数据长时间没有更新就会造成资源的浪费，而且因为设置了多长时间刷新一次，所以数据的显示有延迟。

长轮询

web版的qq或者微信等都是采用长轮询实现的。

原理：服务端将用户的请求夯住，比如夯住10s，如果在这10s中有票数或消息，则立刻返回响应，否则，到指定时间后自动返回响应，然后，客户端自动再次发送请求。

import queue

import uuid

import json

from flask import Flask,request,render_template,make_response,session,jsonify

app = Flask(__name__)

app.secret_key = 'asdf asdf'

USER_DICT = {

    '':{'name':'野味','count':1},

    '':{'name':'海龙','count':1},

}

QUEUE_DICT = {}

@app.route('/index',methods=['GET','POST'])

def index():

    # 为每一个访问页面的用户生成一个唯一的标识

    user_id = str(uuid.uuid4())

    # 为每一个用户生成一个q对象，并添加到全局中，同时在用户的cookie中携带这个唯一标识

    QUEUE_DICT[user_id] = queue.Queue()

    session['user_id'] = user_id

    return render_template('index.html',user_dict=USER_DICT)

@app.route('/get_new_vote')

def get_new_vote():

    result = {'status':True,'data':None}

    # 根据user_id获取当前用户的queue

    user_id = session['user_id']

    q = QUEUE_DICT[user_id]

    try:

        data = q.get(timeout=10)

        result['data'] =data

    except queue.Empty as e:

        result['status'] = False

    # return json.dumps(result)

    return jsonify(result) # JsonResponse

@app.route('/vote',methods=['POST'])

def vote():

    uid = request.form.get('uid')

    USER_DICT[uid]['count'] += 1

    ticket_info = {'uid':uid, 'count':USER_DICT[uid]['count']}

    for q in QUEUE_DICT.values():

        q.put(ticket_info)

    return '投票成功'

if __name__ == '__main__':

    app.run(host='0.0.0.0',debug=False,threaded=True)

备注：uuid.uuid4() 不是json支持的序列化数据类型，所有需要先转化为str

<!DOCTYPE html>

<html lang="en">

<head>

    <meta charset="UTF-8">

    <title>Title</title>

</head>

<body>

    <h1>最丑的男人</h1>

    <ul>

        {% for k,v in user_dict.items() %}

            <li id="user_{{k}}" style="cursor: pointer" ondblclick="vote({{k}})">{{v.name}}<span>{{v.count}}</span></li>

        {% endfor %}

    </ul>

    <script src="https://cdn.bootcss.com/jquery/3.3.1/jquery.min.js"></script>

    <script>

        $(function () {

            get_new_vote();

        });

        function get_new_vote() {

            $.ajax({

                url:'/get_new_vote',

                type:'GET',

                dataType:'json',

                success:function (arg) {

                    if(arg.status){

                        $('#user_'+arg.data.uid).find('span').text(arg.data.count);

                    }

                    get_new_vote();

                }

            })

        }

        function vote(uid) {

            $.ajax({

                url:'/vote',

                type:'POST',

                data:{uid:uid},

                success:function (arg) {

                    console.log(arg);

                }

            })

        }

    </script>

</body>

</html>

HTML

我们借助了队列实现了长轮询，同样的也可以使用redis实现，借助redis的列表也可以实现队列。

备注：redis中的blpop(name,timeout="xx") ，可以设置超时时间。

websocket协议

websocket协议与HTTP协议的最大的区别是：HTTP协议是一次请求和一次响应后断开连接，而websocket是一次请求和一次连接后连接不断开。

由于在flask中默认使用wsgi的模块是werkzeug，而werkzeug是不支持websocket的，所以，使用flask发送websocket请求时，需要借助一个第三方的包----->gevent-websocket

安装： pip install gevent-websocket

gevent-websocket是一个wsgi协议的模块，内部支持http请求，同时也支持websocket请求，所以，使用flask时，要将flask的werkzeug替换掉。

用法：

先导入：

from geventwebsocket.handler import WebSocketHandler
from gevent.pywsgi import WSGIServer

在启动flask时，替换掉原来的启动方式：

http_server = WSGIServer(('0.0.0.0', 5000), app, handler_class=WebSocketHandler)    # 启动传入ip、端口,还有app对象
http_server.serve_forever()

使用websocket发送请求的方式：在js中new一个WebSocket("ws://127.0.0.1:5000/get_vote");页面就会自动发送一个websocket的请求。

备注：websocket中的参数url的格式： ws://ip:port/路径不同于HTTP请求的: http://.... websocket的请求是以ws起头的。

同样的，对于一个路径，既可以接收HTTP协议的请求，也可以接收websocket的请求，所以，在后端，试图函数层就要对请求使用的协议加以区分。

后端视图函数区分websocket和HTTP请求的方法：request.environ.get("wsgi.websocket") ，如果是http请求，返回None，如果是websocket请求，返回一个<geventwebsocket.websocket.WebSocket object at 0x00000232849FE458>

可以借助这个参数加以区分。在后端，可以使用接收到的ws对象 ws=request.environ.get("wsgi.websocket")

发送数据： ws.send("数据") 接收数据： ws.receive()

在前端，可以通过一个回调函数接收数据，通过send发送数据

<script>

    var ws = new WebSocket("ws://127.0.0.1:5000/get_vote");

    //  有消息时，会自动触发执行回调函数

    ws.onmessage = function (ev) {

        console.log(ev.data);

    };

     function ticket(nid){

        ws.send('发送的数据');

    }

</script>

基于websocket实现投票示例：

<!DOCTYPE html>

<html lang="en">

<head>

    <meta charset="UTF-8">

    <title>Title</title>

</head>

<body>

    <h1>最丑的男人</h1>

    <ul>

        {% for k,v in user_dict.items() %}

            <li id="user_{{k}}" style="cursor: pointer" ondblclick="vote({{k}})">{{v.name}}<span>{{v.count}}</span></li>

        {% endfor %}

    </ul>

    <script src="https://cdn.bootcss.com/jquery/3.3.1/jquery.min.js"></script>

    <script>

        // 向后台发送websocekt请求

        var ws = new WebSocket('ws://192.168.12.200:5000/get_new_vote');

        ws.onmessage = function (ev) {

            var ticket = JSON.parse(ev.data);

            $('#user_'+ticket.uid).find('span').text(ticket.count);

        };

        function vote(uid) {

            ws.send(uid)

        }

    </script>

</body>

</html>

HTML

import queue

import uuid

import json

from flask import Flask,request,render_template,make_response,session,jsonify

from geventwebsocket.handler import WebSocketHandler

from gevent.pywsgi import WSGIServer

app = Flask(__name__)

app.secret_key = 'asdf asdf'

USER_DICT = {

    '':{'name':'野味','count':1},

    '':{'name':'海龙','count':1},

}

@app.route('/index',methods=['GET','POST'])

def index():

    return render_template('index.html',user_dict=USER_DICT)

# http://127.0.0.1:5000/get_new_vote

# ws://127.0.0.1:5000/get_new_vote

WEBSOCKET_LIST = []

@app.route('/get_new_vote')

def get_new_vote():

    ws = request.environ.get('wsgi.websocket')

    if not ws:

        return "请使用websocket协议"

    # 浏览器发送的socket客户端

    WEBSOCKET_LIST.append(ws)

    while True:

        uid = ws.receive()

        // 如果用户关闭浏览器，会收到一个空

        if not uid:

            ws.close()

            WEBSOCKET_LIST.remove(ws)

        USER_DICT[uid]['count'] += 1

        # ws.send('666')

        ticket_info = {'uid':uid,'count':USER_DICT[uid]['count']}

        for item in WEBSOCKET_LIST:

            item.send(json.dumps(ticket_info))

if __name__ == '__main__':

    # app.run(host='0.0.0.0',debug=False,threaded=True)

    # app.run(debug=False,threaded=True)

    # http_server = WSGIServer(('127.0.0.1', 5000), app, handler_class=WebSocketHandler)

    http_server = WSGIServer(('0.0.0.0', 5000), app, handler_class=WebSocketHandler)

    http_server.serve_forever()

.py文件

备注：如果用户关闭浏览器，ws.receive()会接收到空，所以要将ws关闭，ws.close().

websocket建立连接时，会先进行握手，认证的过程，

请求和响应的【握手】信息需要遵循规则：

从请求【握手】信息中提取 Sec-WebSocket-Key
利用magic_string 和 Sec-WebSocket-Key 进行hmac1加密，再进行base64加密
将加密结果响应给客户端

注：magic string为：258EAFA5-E914-47DA-95CA-C5AB0DC85B11

发起websocket请求时，会先进行一个认证握手的过程，这个过程为：

1.用户发起websocket请求后，会先在请求头中获取Sec-WebSocket-Key对应的值，再将这个值与majic_string（魔法字符串<这个字符串是固定的>）相加，将相加的结果先通过hashlib.sha1加密，在通过base64加密，然后在将加密后的结果返回给客户端。客户端收到后校验是不是采用的这张加密，通过后，建立起连接，否则就会拒绝连接。

握手成功后才能发送数据，而且收发数据是加密的。

import socket

import base64

import hashlib

def get_headers(data):

    """

    将请求头格式化成字典

    :param data:

    :return:

    """

    header_dict = {}

    data = str(data, encoding='utf-8')

    for i in data.split('\r\n'):

        print(i)

    header, body = data.split('\r\n\r\n', 1)

    header_list = header.split('\r\n')

    for i in range(0, len(header_list)):

        if i == 0:

            if len(header_list[i].split(' ')) == 3:

                header_dict['method'], header_dict['url'], header_dict['protocol'] = header_list[i].split(' ')

        else:

            k, v = header_list[i].split(':', 1)

            header_dict[k] = v.strip()

    return header_dict

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

sock.bind(('127.0.0.1', 8002))

sock.listen(5)

conn, address = sock.accept()

data = conn.recv(1024)

headers = get_headers(data) # 提取请求头信息

# 对请求头中的sec-websocket-key进行加密

response_tpl = "HTTP/1.1 101 Switching Protocols\r\n" \

      "Upgrade:websocket\r\n" \

      "Connection: Upgrade\r\n" \

      "Sec-WebSocket-Accept: %s\r\n" \

      "WebSocket-Location: ws://%s%s\r\n\r\n"

magic_string = '258EAFA5-E914-47DA-95CA-C5AB0DC85B11'

value = headers['Sec-WebSocket-Key'] + magic_string

ac = base64.b64encode(hashlib.sha1(value.encode('utf-8')).digest())

response_str = response_tpl % (ac.decode('utf-8'), headers['Host'], headers['url'])

# 响应【握手】信息

conn.send(bytes(response_str, encoding='utf-8'))

客户端和服务端收发数据

客户端和服务端传输数据时，需要对数据进行【封包】和【解包】。

客户端的JavaScript类库已经封装【封包】和【解包】过程，但Socket服务端需要手动实现。

解包过程：

服务端接收到客户端发送的数据后，后先取出第二个字节的后七位，然后，对这后七位进行一个判断：（后七位转化为数字最大为127），如果后七位<=125，那么前两个字节就是报文；如果后七位=126，则继续往后读16个字节，也就是取前四个字节作为报文；如果后七位=127，则往后读64位，也就是取前10个字节作为报文。然后剩余的数据取前四个字节作为masking_key(掩码key)，对剩下的数据在每个字节的与masking_key做位运算，最终得到真实的数据。

Decoding Payload Length

To read the payload data, you must know when to stop reading. That's why the payload length is important to know. Unfortunately, this is somewhat complicated. To read it, follow these steps:

    Read bits 9-15 (inclusive) and interpret that as an unsigned integer. If it's 125 or less, then that's the length; you're done. If it's 126, go to step 2. If it's 127, go to step 3.

    Read the next 16 bits and interpret those as an unsigned integer. You're done.

    Read the next 64 bits and interpret those as an unsigned integer (The most significant bit MUST be 0). You're done.

Reading and Unmasking the Data

If the MASK bit was set (and it should be, for client-to-server messages), read the next 4 octets (32 bits); this is the masking key. Once the payload length and masking key is decoded, you can go ahead and read that number of bytes from the socket. Let's call the data ENCODED, and the key MASK. To get DECODED, loop through the octets (bytes a.k.a. characters for text data) of ENCODED and XOR the octet with the (i modulo 4)th octet of MASK. In pseudo-code (that happens to be valid JavaScript):

var DECODED = "";

for (var i = 0; i < ENCODED.length; i++) {

    DECODED[i] = ENCODED[i] ^ MASK[i % 4];

}

Now you can figure out what DECODED means depending on your application.

基于python实现websocket的解包过程

import socket

import base64

import hashlib

def get_headers(data):

    """

    将请求头格式化成字典

    :param data:

    :return:

    """

    header_dict = {}

    data = str(data, encoding='utf-8')

    # for i in data.split('\r\n'):

    #     print(i)

    header, body = data.split('\r\n\r\n', 1)

    header_list = header.split('\r\n')

    for i in range(0, len(header_list)):

        if i == 0:

            if len(header_list[i].split(' ')) == 3:

                header_dict['method'], header_dict['url'], header_dict['protocol'] = header_list[i].split(' ')

        else:

            k, v = header_list[i].split(':', 1)

            header_dict[k] = v.strip()

    return header_dict

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

sock.bind(('127.0.0.1', 8002))

sock.listen(5)

conn, address = sock.accept()

data = conn.recv(1024)

headers = get_headers(data)  # 提取请求头信息

# 对请求头中的sec-websocket-key进行加密

response_tpl = "HTTP/1.1 101 Switching Protocols\r\n" \

               "Upgrade:websocket\r\n" \

               "Connection: Upgrade\r\n" \

               "Sec-WebSocket-Accept: %s\r\n" \

               "WebSocket-Location: ws://%s%s\r\n\r\n"

magic_string = '258EAFA5-E914-47DA-95CA-C5AB0DC85B11'

value = headers['Sec-WebSocket-Key'] + magic_string

ac = base64.b64encode(hashlib.sha1(value.encode('utf-8')).digest())

response_str = response_tpl % (ac.decode('utf-8'), headers['Host'], headers['url'])

# 响应【握手】信息

conn.send(bytes(response_str, encoding='utf-8'))

while True:

    info = conn.recv(8096)

    payload_len = info[1] & 127

    if payload_len == 126:

        extend_payload_len = info[2:4]

        mask = info[4:8]

        decoded = info[8:]

    elif payload_len == 127:

        extend_payload_len = info[2:10]

        mask = info[10:14]

        decoded = info[14:]

    else:

        extend_payload_len = None

        mask = info[2:6]

        decoded = info[6:]

    bytes_list = bytearray()

    for i in range(len(decoded)):

        chunk = decoded[i] ^ mask[i % 4]

        bytes_list.append(chunk)

    body = str(bytes_list, encoding='utf-8')

    print(body)

python实现解包

封包过程：

向客户端发送数据【封包】

def send_msg(conn, msg_bytes):

    """

    WebSocket服务端向客户端发送消息

    :param conn: 客户端连接到服务器端的socket对象,即： conn,address = socket.accept()

    :param msg_bytes: 向客户端发送的字节

    :return:

    """

    import struct

    token = b"\x81"

    length = len(msg_bytes)

    if length < 126:

        token += struct.pack("B", length)

    elif length <= 0xFFFF:

        token += struct.pack("!BH", 126, length)

    else:

        token += struct.pack("!BQ", 127, length)

    msg = token + msg_bytes

    conn.send(msg)

    return True

基于python实现封包

基于python socket实现WebSocket服务端

import socket

import base64

import hashlib

def get_headers(data):

    """

    将请求头格式化成字典

    :param data:

    :return:

    """

    header_dict = {}

    data = str(data, encoding='utf-8')

    header, body = data.split('\r\n\r\n', 1)

    header_list = header.split('\r\n')

    for i in range(0, len(header_list)):

        if i == 0:

            if len(header_list[i].split(' ')) == 3:

                header_dict['method'], header_dict['url'], header_dict['protocol'] = header_list[i].split(' ')

        else:

            k, v = header_list[i].split(':', 1)

            header_dict[k] = v.strip()

    return header_dict

def send_msg(conn, msg_bytes):

    """

    WebSocket服务端向客户端发送消息

    :param conn: 客户端连接到服务器端的socket对象,即： conn,address = socket.accept()

    :param msg_bytes: 向客户端发送的字节

    :return:

    """

    import struct

    token = b"\x81"

    length = len(msg_bytes)

    if length < 126:

        token += struct.pack("B", length)

    elif length <= 0xFFFF:

        token += struct.pack("!BH", 126, length)

    else:

        token += struct.pack("!BQ", 127, length)

    msg = token + msg_bytes

    conn.send(msg)

    return True

def run():

    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

    sock.bind(('127.0.0.1', 8003))

    sock.listen(5)

    conn, address = sock.accept()

    data = conn.recv(1024)

    headers = get_headers(data)

    response_tpl = "HTTP/1.1 101 Switching Protocols\r\n" \

                   "Upgrade:websocket\r\n" \

                   "Connection:Upgrade\r\n" \

                   "Sec-WebSocket-Accept:%s\r\n" \

                   "WebSocket-Location:ws://%s%s\r\n\r\n"

    value = headers['Sec-WebSocket-Key'] + '258EAFA5-E914-47DA-95CA-C5AB0DC85B11'

    ac = base64.b64encode(hashlib.sha1(value.encode('utf-8')).digest())

    response_str = response_tpl % (ac.decode('utf-8'), headers['Host'], headers['url'])

    conn.send(bytes(response_str, encoding='utf-8'))

    while True:

        try:

            info = conn.recv(8096)

        except Exception as e:

            info = None

        if not info:

            break

        payload_len = info[1] & 127

        if payload_len == 126:

            extend_payload_len = info[2:4]

            mask = info[4:8]

            decoded = info[8:]

        elif payload_len == 127:

            extend_payload_len = info[2:10]

            mask = info[10:14]

            decoded = info[14:]

        else:

            extend_payload_len = None

            mask = info[2:6]

            decoded = info[6:]

        bytes_list = bytearray()

        for i in range(len(decoded)):

            chunk = decoded[i] ^ mask[i % 4]

            bytes_list.append(chunk)

        body = str(bytes_list, encoding='utf-8')

        send_msg(conn,body.encode('utf-8'))

    sock.close()

if __name__ == '__main__':

    run()

基于python

基于Javascript实现客户端

<!DOCTYPE html>

<html>

<head lang="en">

    <meta charset="UTF-8">

    <title></title>

</head>

<body>

    <div>

        <input type="text" id="txt"/>

        <input type="button" id="btn" value="提交" onclick="sendMsg();"/>

        <input type="button" id="close" value="关闭连接" onclick="closeConn();"/>

    </div>

    <div id="content"></div>

<script type="text/javascript">

    var socket = new WebSocket("ws://127.0.0.1:8003/chatsocket");

    socket.onopen = function () {

        /* 与服务器端连接成功后，自动执行 */

        var newTag = document.createElement('div');

        newTag.innerHTML = "【连接成功】";

        document.getElementById('content').appendChild(newTag);

    };

    socket.onmessage = function (event) {

        /* 服务器端向客户端发送数据时，自动执行 */

        var response = event.data;

        var newTag = document.createElement('div');

        newTag.innerHTML = response;

        document.getElementById('content').appendChild(newTag);

    };

    socket.onclose = function (event) {

        /* 服务器端主动断开连接时，自动执行 */

        var newTag = document.createElement('div');

        newTag.innerHTML = "【关闭连接】";

        document.getElementById('content').appendChild(newTag);

    };

    function sendMsg() {

        var txt = document.getElementById('txt');

        socket.send(txt.value);

        txt.value = "";

    }

    function closeConn() {

        socket.close();

        var newTag = document.createElement('div');

        newTag.innerHTML = "【关闭连接】";

        document.getElementById('content').appendChild(newTag);

    }

</script>

</body>

</html>