当前位置：首页 > news >正文

sanic + webSocket：股票实时行情推送服务实现

news 文章来源：https://blog.csdn.net/sxc1414749109/article/details/141222196 2025/5/5 20:00:18

💝💝💝欢迎莅临我的博客，很高兴能够在这里和您见面！希望您在这里可以感受到一份轻松愉快的氛围，不仅可以获得有趣的内容和知识，也可以畅所欲言、分享您的想法和见解。
在这里插入图片描述

推荐：「stormsha的主页」👈，「stormsha的知识库」👈持续学习，不断总结，共同进步，为了踏实，做好当下事儿~
专栏导航
- Python系列: Python面试题合集，剑指大厂
- Git系列: Git操作技巧
- GO系列: 记录博主学习GO语言的笔记，该笔记专栏尽量写的试用所有入门GO语言的初学者
- 数据库系列: 详细总结了常用数据库 mysql 技术点,以及工作中遇到的 mysql 问题等
- 运维系列: 总结好用的命令，高效开发
- 算法与数据结构系列: 总结数据结构和算法,不同类型针对性训练,提升编程思维
非常期待和您一起在这个小小的网络世界里共同探索、学习和成长。💝💝💝 ✨✨ 欢迎订阅本专栏 ✨✨

💖The Start💖点点关注，收藏不迷路💖
📒文章目录
- - 1、Sanic WebSocket 实现股票实时推送服务
  - - 1.1. 设置 Sanic WebSocket 服务端
    - 1.2. 创建 HTML 客户端代码
  - 2. 实现客户端与服务端的交互
  - 3. 总结

1、Sanic WebSocket 实现股票实时推送服务

在本文中，我们将详细介绍如何使用 Sanic 框架来实现一个股票实时推送服务，并且展示如何在客户端通过 HTML 和 JavaScript 代码接收这些实时更新。Sanic 是一个异步的 Python 框架，非常适合构建高性能的 Web 应用程序和 API。我们将通过以下几个步骤来实现这个系统：

设置 Sanic WebSocket 服务端
创建 HTML 客户端代码
实现客户端与服务端的交互

1.1. 设置 Sanic WebSocket 服务端

首先，我们需要安装 Sanic 和相关的 WebSocket 库。可以使用以下命令来安装：

pip install sanic sanic_cors websockets

接下来，我们创建一个简单的 Sanic 应用程序，并使用 WebSocket 实现实时的股票数据推送。

import json
import uuid
from json import JSONDecodeErrorfrom sanic import Sanic, response
from sanic_cors import CORS, cross_origin
from sanic.websocket import WebSocketProtocol
from websockets import ConnectionClosedError, ConnectionClosedOK
import asyncio
import time
import randomapp = Sanic("StockWebSocket")
CORS(app)# 以股票代码为键的字典，存储订阅者列表
subscriptions = {}# 模拟股票行情数据，实际开发中用真实行情源代替
stock_data = {"a": {"stock_name": "阿里", "price": 150.00, "change": 0.25},"b": {"stock_name": "百度", "price": 2750.00, "change": 0.25},"t": {"stock_name": "腾讯", "price": 3400.00, "change": 0.25}
}class SubscriptObject(object):def __init__(self, register_id, ws):self.register_id = register_idself.ws = wsasync def recv(self):try:while True:msg = await self.ws.recv()if not msg:continuedata = json.loads(msg)action = data.get('action')stock_symbol = data.get('stock_symbol')if action == "subscribe":if stock_symbol in stock_data:if not subscriptions.get(stock_symbol):subscriptions[stock_symbol] = set()subscriptions[stock_symbol].add(self.ws)else:await self.ws.send(json.dumps({"error": "Unknown stock symbol"}))else:print(f"当前状态 {data}")except ConnectionClosedError:print(f'{self.register_id} 断开连接, 取消监听')except ConnectionClosedOK:print(f'{self.register_id} 断开连接, 取消监听')except JSONDecodeError:print(f'收到的数据解析失败')finally:# 清理客户端订阅for symbol in subscriptions:subscriptions[symbol].discard(self.ws)print("Client disconnected")async def stock_data_generator():while True:# 模拟股票数据更新stock_code_list = list(subscriptions.keys())for symbol in stock_code_list:ws_list = subscriptions.get(symbol, []).copy()dd = stock_data.get(symbol)data = dict()data['stock_name'] = dd['stock_name']data['price'] = dd['price'] + random.uniform(-5, 5)data['change'] = dd['change'] + random.uniform(-1, 1)data['timestamp'] = time.time()for ws in ws_list:try:await ws.send(json.dumps(data))except ConnectionClosedOK:print("连接取消前", subscriptions[symbol])subscriptions[symbol].discard(ws)print("连接取消后", subscriptions[symbol])await asyncio.sleep(1)  # 每2秒更新一次@app.websocket("/ws/stock")
@cross_origin(app)  # 为这个路由启用CORS
async def stock_websocket(request, ws):# 模拟实时更新股票价格register_id = uuid.uuid4().hexsub_obj = SubscriptObject(register_id, ws)await sub_obj.recv()# 启动股票数据更新任务
app.add_task(stock_data_generator())@app.route('/')
async def index(request):return await response.file('index.html')if __name__ == "__main__":app.run(host="0.0.0.0", port=8000, protocol=WebSocketProtocol)

1.2. 创建 HTML 客户端代码

我们需要一个简单的 HTML 页面，允许用户选择股票符号，并通过 WebSocket 连接到服务端以接收实时更新。

<!DOCTYPE html>
<html lang="en">
<head><meta charset="UTF-8"><meta name="viewport" content="width=device-width, initial-scale=1.0"><title>Stock Real-time Update</title><style>body {font-family: Arial, sans-serif;}.stock-data {margin: 20px;padding: 10px;border: 1px solid #ddd;}select {margin: 20px;padding: 5px;}</style>
</head>
<body>
<h1>实时股票行情</h1>
<select id="stock-select"><option value="a">阿里</option><option value="b">百度</option><option value="t">腾讯</option>
</select>
<div id="stock-container"><!-- 初始化一个空的 stock-data 元素，用于动态添加新的 stock-data 元素 --><div class="stock-data" id="stock-data-template" style="display: none;">股票名称: <span class="stock-name"></span><br>涨跌幅: <span class="change"></span><br>时间戳: <span class="timestamp"></span><br></div>
</div><script>const stockSelect = document.getElementById('stock-select');const stockContainer = document.getElementById('stock-container');const stockDataTemplate = document.getElementById('stock-data-template');function updateStock(symbol) {const ws = new WebSocket(`ws://localhost:8000/ws/stock`);const stockDataElement = stockDataTemplate.cloneNode(true); // 复制 stock-data 模板元素stockDataElement.id = `stock-data-${symbol}`; // 根据股票代码设置 IDstockDataElement.style.display = 'block'; // 显示元素stockContainer.appendChild(stockDataElement); // 将元素添加到容器中ws.onopen = function () {console.log("WebSocket connection established.");ws.send(JSON.stringify({action: 'subscribe',stock_symbol: symbol}));};ws.onmessage = function (event) {const data = JSON.parse(event.data);const stockData = stockDataElement.getElementsByClassName('stock-data')[0];if (data.error) {stockData.innerHTML = `<p>${data.error}</p>`;} else {console.log("ddddddd", data)stockDataElement.getElementsByClassName('stock-name')[0].textContent = data.stock_name;stockDataElement.getElementsByClassName('change')[0].textContent = data.change;stockDataElement.getElementsByClassName('timestamp')[0].textContent = data.timestamp;}};ws.onerror = function (error) {console.log("WebSocket Error: ", error);};}stockSelect.addEventListener('change', function () {const selectedSymbol = stockSelect.value;updateStock(selectedSymbol);});// 初始化加载第一个股票的数据updateStock(stockSelect.value);
</script>
</body>
</html>

访问地址：http://localhost:8000

2. 实现客户端与服务端的交互

在上面的代码中，客户端通过 WebSocket 连接到服务端，并根据输入的股票符号接收实时更新。服务端定期推送模拟的股票数据到所有订阅该股票符号的客户端。

3. 总结

通过以上步骤，我们实现了一个基于 Sanic 的 WebSocket 股票实时推送服务，并创建了一个简单的 HTML 客户端以接收这些实时更新。这个示例可以根据实际需求进行扩展和优化，如连接管理、数据来源的整合、错误处理等。希望这个示例能帮助你在实际项目中实现实时数据推送功能。

🔥🔥🔥道阻且长,行则将至,让我们一起加油吧！🌙🌙🌙

💖The End💖点点关注，收藏不迷路💖

sanic + webSocket：股票实时行情推送服务实现

📒文章目录

1、Sanic WebSocket 实现股票实时推送服务

1.1. 设置 Sanic WebSocket 服务端

1.2. 创建 HTML 客户端代码

2. 实现客户端与服务端的交互

3. 总结

相关文章：

sanic + webSocket：股票实时行情推送服务实现

Unity动态给按钮各个状态下的图片赋值

xiaomi pad 6PRO 小米平板6 pro hyperOS降级澎湃os 降级MIUI 14 教程免解锁BL 降级，168小时解锁绑定

MySQL 备份一个表

鸿蒙开发入门day10-组件导航

虚拟机Linux的坑 | VMware无法从主机向虚拟机跨系统复制粘贴拖动文件/文本

Chat App 项目之解析（二）

数据结构与算法 - 双指针

Python3网络爬虫开发实战（10）模拟登录（需补充账号池的构建）

SQL 调优最佳实践笔记

Eclipse的使用配置教程：必要设置、创建工程及可能遇到的问题(很详细，很全面，能解决90%的问题)

遗传算法与深度学习实战（4）——遗传算法详解与实现

Nginx+Tomcat实现负载均衡、动静分离集群部署

英语学习8月19日

关于windows环境使用nginx的一些性能问题

“解决Windows电脑无法投影到其他屏幕的问题：尝试更新驱动程序或更换视频卡“

第10章无持久存储的文件系统（2）

云计算实训29——mysql主从复制同步、mysql5.7版本安装配置、python操作mysql数据库、mycat读写分离实现

AI搜索引擎Perplexica的本地部署（之二）Perplexica的非docker安装

Oracle环境下在相同参数和数据源的情况,mybatis-plus查询和sql查询结果不一致

springboot静态资源访问问题归纳

HTML与CSS学习Day01

Tina-Linux Bootloaer简述

【Python】 Scrapyd：Python Web Scraping 的强大分布式调度工具

吴恩达机器学习课后题-01线性回归

白盒报告-jacoco

【MySQL】SQL语句执行流程

Selenium自动化防爬技巧：从入门到精通，保障爬虫稳定运行，通过多种方式和add_argument参数设置来达到破解防爬的目的

从数据类型到变量、作用域、执行上下文

一文读懂：AI时代到底需要什么样的网络？