Sanic 长轮询实现股票行情实时更新

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

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

💖The Start💖点点关注，收藏不迷路💖
📒文章目录
- - 1. 安装依赖包
  - 2. Sanic 服务器端代码
  - 3. 网页客户端代码
  - 4. 代码解析
  - 5. 启动服务器和测试

1. 安装依赖包

pip install sanic
pip install sanic-cors

2. Sanic 服务器端代码

首先，设置 Sanic 服务器来处理股票数据的长轮询请求。服务器将根据用户选择的股票代码返回对应的股票数据。

python">from sanic import Sanic, response
from sanic_cors import CORS, cross_origin
import asyncio
import timeapp = Sanic("StockLongPollingApp")# 启用 CORS
CORS(app)# 模拟的股票行情数据
stock_data = {"A": {"stock_name": "Apple", "price": 150.00, "change": 0.25},"B": {"stock_name": "Google", "price": 2750.00, "change": -10.00},"T": {"stock_name": "Amazon", "price": 3400.00, "change": 15.00}
}# 模拟股票数据更新的任务
async def update_stock_data():while True:# 更新股票价格 (这里只是简单模拟实际应用中可能来自数据库或外部API)for symbol in stock_data:stock_data[symbol]["price"] += (time.time() % 10 - 5) * 0.1stock_data[symbol]["change"] = (time.time() % 2 - 1) * 0.1await asyncio.sleep(10)@app.route("/poll/<stock_symbol>")
@cross_origin(app)  # 为这个路由启用CORS
async def poll_stock(request, stock_symbol):timeout = 30  # 长轮询的超时时间async def wait_for_event():start_time = time.time()last_data = Nonewhile True:# 获取当前股票数据current_data = stock_data.get(stock_symbol, None)if current_data and current_data != last_data:last_data = current_datareturn current_dataelif time.time() - start_time > timeout:return Noneawait asyncio.sleep(1)data = await wait_for_event()if data:return response.json(data)else:return response.json({"message": "No new data"}, status=204)if __name__ == "__main__":# 启动模拟股票数据更新的任务asyncio.ensure_future(update_stock_data())app.run(host="0.0.0.0", port=8000)

3. 网页客户端代码

网页客户端允许用户选择股票，并通过长轮询机制实时获取选定股票的行情数据。

<!DOCTYPE html>
<html lang="en">
<head><meta charset="UTF-8"><meta name="viewport" content="width=device-width, initial-scale=1.0"><title>Stock Price Long Polling</title><style>body {font-family: Arial, sans-serif;margin: 20px;}#data {margin-top: 20px;padding: 10px;border: 1px solid #ddd;}select {padding: 5px;margin-right: 10px;}</style>
</head>
<body><h1>Real-time Stock Prices</h1><label for="stock-select">Choose a stock:</label><select id="stock-select"><option value="A">阿里</option><option value="B">百度</option><option value="T">腾讯</option></select><div id="data">Select a stock to see real-time data...</div><script>let currentStock = 'A';document.getElementById('stock-select').addEventListener('change', function() {currentStock = this.value;fetchData();  // 立即触发新股票的数据获取});async function fetchData() {try {const response = await fetch(`http://localhost:8000/poll/${currentStock}`);if (response.ok) {const result = await response.json();document.getElementById('data').innerText = `Stock: ${result.stock_name}, Price: $${result.price.toFixed(2)}, Change: ${result.change.toFixed(2)}`;} else if (response.status === 204) {document.getElementById('data').innerText = 'No new data';}} catch (error) {console.error('Error fetching data:', error);}setTimeout(fetchData, 1000);}fetchData();</script>
</body>
</html>

4. 代码解析

服务器端：
- stock_data：模拟股票行情数据，每10秒更新一次。
- poll_stock：处理长轮询请求，检查是否有新数据可用，超时返回204状态码。
- update_stock_data：异步任务模拟股票价格的变化。
客户端：
- 选择股票：用户通过 select 元素选择要查看的股票。
- 长轮询请求：fetchData 函数使用 fetch API 每秒发起一次请求，更新所选股票的实时数据。
- 数据展示：获取到新数据后，更新 div 元素中的内容。