websockets使用介绍（RPC调用）

我们想要实现在浏览器环境下面获取某些数据，然后在python端进行使用。可以考虑通过websockets来实现。

一、首先要用python创建一个websockets服务。实例代码如下：

websockets服务，可以理解为中间商，主要的作用就是做代理转发，将浏览器发送过来的消息转发给python客户端，将python客户端发送过来的消息转发给浏览器。

首先需要安装websockets库 pip install websockets

# pip install websockets
import websockets
import asyncio
import re

browser_info = {}
client_info = {}
heart_info = {}


async def regist(ws, path):
    # 注意正则最后了. 如果用.*? 会什么都匹配不到
    obj = re.compile(r"ws/(?P<action>.*?)\.ws\?name=(?P<name>.*)")
    search_result = obj.search(path)
    action = search_result.group("action")
    name = search_result.group("name")
    print("action===",action)
    print("name===",name)
    if action == "regist":  # 来自浏览器
        browser_info[name] = ws  # 保存该链接
        return "browser", name

    elif action == 'invoke':
        client_info[name] = ws
        return "client", name

    elif action =='heartbeat':
        heart_info[name]=ws
        return "heart",name



# 第一个参数ws表示服务器与客户端的链接
# path表示请求过来的路径
async def handle(ws, path):
    # 建立连接的时候走这里.
    t, name = await regist(ws, path)
    async for msg in ws:
        if t == 'browser'and msg not in "ping":
            print("msg==",msg)
            # 接受到的消息来自于浏览器
            await client_info[name].send(msg)

        elif t == 'browser'and msg  in "ping":
            # print("msg==", msg)
            # 接受到的消息来自于浏览器的心跳
            await browser_info[name].send("pong")

        elif t == 'client':
            await browser_info[name].send(msg)
            
        elif t=='heart':
            await heart_info[name].send(msg+"我已收到！")



async def main():
    # 启动websocket服务端
    async with websockets.serve(handle, "127.0.0.1", 8848,ping_interval=None) as ws:
        print("你成功的启动了一个websocket")
        await asyncio.Future()  # 永远停在这里.


if __name__ == '__main__':
    asyncio.run(main())

二、在浏览器端创建一个连接，实例代码如下：

以下两点需要注意：

1、注意考虑和服务器保持连接，通过不断的向服务器发送心跳，防止浏览器和服务器端断开连接。

2、有些时候，我们需要用到的某些值需要刷新浏览器后重新获取，比如有些cookie有有效期，这个时候就要刷新页面重新获取，但是刷新页面会把代码也清理掉。在网上找了很多方法，最后终于实现了，刷新当前页面，但是浏览器的js代码不会被干掉。

详见如下实例代码

(function () {

    var lockReconnect = false;  // 避免ws重复连接
    var ws = null;          // 判断当前浏览器是否支持WebSocket
    var wsUrl = 'ws://xx.xx.xx.xx/ws/regist.ws?name=xxx';
    // 连接ws
    createWebSocket(wsUrl);

    /**
     * 创建 WS 实例
     * @param {string} url ws的URL
     */
    function createWebSocket(url) {
        try {
            if ('WebSocket' in window) {
                ws = new WebSocket(url);
            }
            initEventHandle();
        } catch (e) {
            reconnect(url);
            console.log(e);
        }
    }

    /**
     * 初始化事件处理
     */
    function initEventHandle() {
        ws.onclose = function (e) {
            reconnect(wsUrl);
            console.log("WS 连接关闭!" + +e.code + ' ' + e.reason + ' ' + e.wasClean + new Date().toLocaleString());
            console.log(e)
        };
        ws.onerror = function () {
            reconnect(wsUrl);
            console.log("WS 连接错误!");
        };
        ws.onopen = function () {
            //心跳检测重置
            heartCheck.reset().start();
            console.log("WS 连接成功!" + new Date().toLocaleString());
        };
        ws.onmessage = function (event) {
            //如果获取到消息，心跳检测重置
            //拿到任何消息都说明当前连接是正常的
            heartCheck.reset().start();
            // console.log("WS 收到消息啦:" + event.data);

            // 当接受到消息后. 进行各种计算.
            // let ret = window.qiaofu(); // 你得找到加密入口...
            // 返回结果
            // 当服务器返回的消息不是pong的时候
            if (event.data != 'pong') {
                // let data = JSON.parse(event.data);
                let ret_obj = {
                    cookie: document.cookie
                }
                console.log("计算完毕, 结果是", ret_obj);
                ws.send(JSON.stringify(ret_obj));
            }
        };
    }

// 监听窗口关闭事件，当窗口关闭时，主动去关闭websocket连接，防止连接还没断开就关闭窗口，server端会抛异常。
    window.onbeforeunload = function () {
        ws.close();
    }

    /**
     * 重连
     * @param {string} url ws的URL
     * @returns null
     */
    function reconnect(url) {
        if (lockReconnect) return;
        lockReconnect = true;
        // 没连接上，会一直重连，设置延迟 2s 避免连接频繁
        setTimeout(function () {
            createWebSocket(url);
            lockReconnect = false;
        }, 2000);
    }

    function is_time(qick_time) {
        var d = new Date();
        var year = d.getFullYear();
        var month = d.getMonth() + 1;
        var day = d.getDate();

        var hour = d.getHours();
        var min = d.getMinutes();
        var sec = d.getSeconds();
        // console.log(year, "-", month, "-", day, "-", hour, "-", min, "-", sec)

        curtime = hour.toString() + min.toString()

        // console.log(curtime)
        if (curtime == qick_time) {

            return true
        } else {

            return false
        }

    }

    function reload() {

        current = location.href;
        var frame = '<frameset cols="*">\n<frame src="' + current + '" /></frameset>';
        with (document) {
            // 引用document对象，调用write方法写入框架，打开新窗口
            write(frame);

            //此处输入执行的代码

            // 关闭上面的窗口
            void (close());
        }
        ;
    }


// 心跳检测
    var heartCheck = {
        timeout: 3000,
        timeoutObj: null,
        serverTimeoutObj: null,
        reset: function () {
            clearTimeout(this.timeoutObj);
            clearTimeout(this.serverTimeoutObj);
            return this;
        },
        start: function () {
            var self = this;
            this.timeoutObj = setTimeout(function () {
                //这里发送一个心跳，后端收到后，返回一个心跳消息，
                //onmessage拿到返回的心跳就说明连接正常
                ws.send("ping");
                // console.log("客户端发送了一个心跳ping!")
                //如果超过一定时间还没重置，说明后端主动断开了
                self.serverTimeoutObj = setTimeout(function () {
                    //如果onclose会执行reconnect，我们执行ws.close()就行了.如果直接执行reconnect 会触发onclose导致重连两次
                    ws.close();
                }, self.timeout)
            }, this.timeout)
            // 每天凌晨1点11分刷新一下页面，重新获取一下cookie
            if (is_time("111")) {
                reload()
            }
        }
    }


})()

三、python客户端向浏览器获取数据

以下是简单的python的websockets客户端。

# pip install websockets
import websockets
import asyncio


async def main():
    # python这边链接是为了什么? 为了让ws调用js. 完成加密
    # Python连接websocket服务器的逻辑
    async with websockets.connect("ws://127.0.0.1:8848/ws/invoke.ws?name=xxx") as ws:
    
        await ws.send("你好!")  # 发送数据过去
        print("连接成功了")
        ret = await ws.recv()
        print(ret)


if __name__ == '__main__':
    asyncio.run(main())

在实际使用中，一般用sanic做一个服务（为啥选用sanic是因为sanic支持异步 ），将提供给第三方调用。

from sanic import Sanic, HTTPResponse
import websockets
import asyncio
app = Sanic(__name__)


@app.route("/get")
async def func(req):
    # 在这里. 你可以接受参数. 指定哪个项目
    project_name = req.args.get("project_name")
    if project_name:
        async with websockets.connect(f"ws://127.0.0.1:8848/ws/invoke.ws?name={project_name}") as ws:
            await ws.send("你好")  # 发送数据过去
            print("连接成功了")
            ret = await ws.recv()
        return HTTPResponse(ret)
    else:
        return HTTPResponse("至少要给我一个project_name")


if __name__ == '__main__':
    app.run()

如下是带参数的实例代码：

from sanic import Sanic, HTTPResponse
import websockets
import asyncio
app = Sanic(__name__)
import json


@app.route("/get", methods=["GET", "POST"])
async def func(req):
    # 在这里. 你可以接受参数. 指定哪个项目
    # 在url上面传参过来
    project_name = req.args.get("project_name")
    # 我制定的规则是:
    #   url上面传递的参数: project_name.
    #   该项目需要的其他参数, 全部通过json的形式传递过来
    # 强制规定了. 有参数必须走post. 参数通过json传递过来....
    # 此时浏览器上就无法完成测试了....
    project_params = req.json  # {url: "xxxxx"}
    
    if not project_params:
        project_params = "没参数"

    if project_name:
        async with websockets.connect(f"ws://127.0.0.1:8848/invoke.ws?name={project_name}") as ws:
            await ws.send(json.dumps(project_params))  # 发送数据过去
            print("连接成功了")
            ret = await ws.recv()
        return HTTPResponse(ret)
    else:
        return HTTPResponse("至少要给我一个project_name")


if __name__ == '__main__':
    app.run()

实际过程中，是将websockets服务器端和sanic做的服务（python客户端）部署在Linux服务上。然后第三方调用sanic服务来使用。

第三方调用服务实例代码如下：

import requests

url = "http://127.0.0.1:8000/get?project_name=xxx"
resp = requests.get(url)
print(resp.text)