实时监听数据示例

Posted 2020-11-06 无名！

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了实时监听数据示例相关的知识，希望对你有一定的参考价值。

1、基于长轮询、queue实现

app01.py

#基于queue实现在线实时投票

import uuid
import queue

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

app=Flask(__name__)
app.secret_key="drgs"
app.debug=True

@app.before_request
def check_login():
    if request.path=="/login":
        return None
    user = session.get("user_info")

    if not user:
        return redirect("/login")

USER_QUEUE={

}
"""
USER_QUEUE={
    uid:[data]
}
"""

@app.route("/login",methods=["GET","POST"])
def login():
    if request.method=="GET":
        return render_template("login.html")
    else:
        user=request.form.get("user")
        pwd=request.form.get("pwd")
        if user=="cao" and pwd=="123":
            uid=str(uuid.uuid4())
            #创建队列
            USER_QUEUE[uid]=queue.Queue()
            session["user_info"]={"uid":uid,"name":user}
            return redirect("/index")
        else:
            return render_template("login.html",msg="用户名或密码错误")

VOTE_USER_LIST={
    "1":{"name":"曹超","count":0},
    "2":{"name":"大仙","count":0},
}

@app.route("/index")
def index():
    return render_template("index.html",vote_user_list=VOTE_USER_LIST)



@app.route("/get_vote_count")
def get_vote_count():
    """
    根据用户uid，获取用户队列
    :return:
    """
    ret = {"status":True,"data":None}
    #用户唯一标识
    uid=session["user_info"]["uid"]

    q=USER_QUEUE[uid]
    try:
        data=q.get(timeout=10)
        ret["data"]=data
    except queue.Empty as e:
        ret["status"]=False
    return jsonify(ret)


@app.route("/vote",methods=["POST"])
def vote():
    # 投票对象的id
    gid=request.form.get(‘gid‘)
    old=VOTE_USER_LIST[gid]["count"]
    new=old+1
    VOTE_USER_LIST[gid]["count"]=new

    data={"gid":gid ,"count":new}
    for q in USER_QUEUE.values():
        q.put(data)
    return "ok"



if __name__=="__main__":
    app.run(host="0.0.0.0",threaded=True)

View Code

templates

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta http-equiv="X-UA-Compatible" content="IE=edge">
    <meta name="viewport" content="width=device-width, initial-scale=1">
    <title>登录页面</title>

</head>
<body>

<form method="post">
    <input type="text" name="user">
    <input type="password" name="pwd">
    <input type="submit" value="登录">{{seg}}
</form>



</body>
</html>

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta http-equiv="X-UA-Compatible" content="IE=edge">
    <meta name="viewport" content="width=device-width, initial-scale=1">
    <title>投票页面</title>

</head>
<body>

<h3>选出最具潜力股的男人</h3>

<ul>
    {% for k,v in vote_user_list.items()%}
        <li style="cursor: pointer"    id="{{k}}" ondblclick="vote({{k}})">ID:{{k}} &nbsp姓名:{{v.name}} &nbsp票数:   <span>{{v.count}}</span></li>
    {% endfor %}
</ul>

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

<script>
    $(function () {
        get_vote_count()
    });

    function get_vote_count() {
         $.ajax({
            url:"/get_vote_count",
            type:"GET",
            dataType:‘JSON‘,
        success:function (arg) {
            if (arg.status){
                //更新票数
                var gid="#"+arg.data.gid;
                $(gid).find(‘span‘).text(arg.data.count);
            }else{
                //10s内没有人投票
            }
            get_vote_count()
        }
    });
    }

    //投票
    function vote(gid) {
        $.ajax({
            url:"/vote",
            type:"POST",
            data:{"gid":gid},
            dateType:"JSON",
            success:function (arg) {

            }
        })

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

index.html

2、基于长轮询、redis实现

app02.py

#基于redis实现在线实时投票

import uuid
import redis
import json

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

app=Flask(__name__)
app.secret_key="drgs"
app.debug=True

conn=redis.Redis(host="127.0.0.1",port=6379)

@app.before_request
def check_login():
    if request.path=="/login":
        return None
    user = session.get("user_info")

    if not user:
        return redirect("/login")


@app.route("/login",methods=["GET","POST"])
def login():
    if request.method=="GET":
        return render_template("login.html")
    else:
        user=request.form.get("user")
        pwd=request.form.get("pwd")
        if user=="cao" and pwd=="123":
            uid=str(uuid.uuid4())
            # 放入缓存中
            conn.lpush("uuid",uid)
            session["user_info"]={"uid":uid,"name":user}
            return redirect("/index")
        else:
            return render_template("login.html",msg="用户名或密码错误")

VOTE_USER_LIST={
    "1":{"name":"曹超","count":0},
    "2":{"name":"大仙","count":0},
}

@app.route("/index")
def index():
    return render_template("index.html",vote_user_list=VOTE_USER_LIST)


@app.route("/get_vote_count")
def get_vote_count():
    """
    根据用户uid，获取用户队列
    :return:
    """
    ret = {"status":True,"data":None}
    #用户唯一标识
    uid=session["user_info"]["uid"]

    data=conn.brpop(uid,timeout=10)
    if not data:
        ret["status"]=False
    else:
        # 数据格式转化
        data_str=str(data[1],encoding="utf-8")
        data_dict=json.loads(data_str)
        ret["data"]=data_dict
    return jsonify(ret)


@app.route("/vote",methods=["POST"])
def vote():
    # 投票对象的id
    gid=request.form.get(‘gid‘)
    old=VOTE_USER_LIST[gid]["count"]
    new=old+1
    VOTE_USER_LIST[gid]["count"]=new
    #传入字节格式
    data=json.dumps({"gid":gid ,"count":new})

    #获取uuid
    uid_list = conn.lrange(‘uuid‘, 0, conn.llen(‘uuid‘))
    for uid in uid_list:
        conn.lpush(uid, data)
    return "ok"



if __name__=="__main__":
    app.run(host="127.0.0.1",port=4000,threaded=True)

View Code

templates一样

3、基于websocket

text.py

#基于websocket实现实时监听数据

from geventwebsocket.handler import WebSocketHandler
from gevent.pywsgi import WSGIServer
from flask import Flask,render_template,request,session,redirect,jsonify
import pickle

app=Flask(__name__)


@app.route("/text")
def text():
    return render_template("text.html")

WS_LIST=[]
@app.route("/get_text")
def get_text():
    ws=request.environ.get("wsgi.websocket")
    if not ws:
        print("请使用Websocket协议规范")
    # 此时websocket连接创建成功
    WS_LIST.append(ws)

    while True:
        # 接受数据
        message=ws.receive()

        # 关闭
        if not message:
            WS_LIST.remove(ws)
            ws.colse()
            break

        # 发送数据
        for item in WS_LIST:
            item.send(message)

    return "ok"

if __name__=="__main__":
    http_server=WSGIServer(("0.0.0.0",5000),app,handler_class=WebSocketHandler)
    http_server.serve_forever()

View Code

text.html

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta http-equiv="X-UA-Compatible" content="IE=edge">
    <meta name="viewport" content="width=device-width, initial-scale=1">
    <title>Title</title>

</head>
<body>

<h3>发送消息</h3>
<input id="msg" type="text" >
<input type="submit" valut="发送" onclick="sendmsg()">

<div id="message-list">
    
</div>

<script src="/static/jquery-3.3.1.min.js"></script>
<script>
    <!--连接客户端-->
    ws= new WebSocket("ws://192.168.21.40:5000/get_text");
    ws.onmessage=function (arg) {
        
        var tag=document.createElement("div");
        tag.className="msg-itm";
        tag.innerText=arg.data;
        $("#message-list").append(tag)
    };
    function sendmsg() {
        ws.send($("#msg").val())
        
    }


</script>


</body>
</html>

View Code

以上是关于实时监听数据示例的主要内容，如果未能解决你的问题，请参考以下文章

监听网络改变, 实时获取网络连通性,ping/发请求

永远观察实时数据的片段

如何实现一个 Firebase 监听器来观察数据库的实时变化

js/jquery 实时监听输入框值变化的完美方案：oninput & onpropertychange

实时数据中的 Observables 更新在片段中不起作用

检索数据未出现在 ListView 的片段中