python 基于FastAPI实现一个简易的在线用户统计 服务

简易在线用户统计服务

概述

这是一个基于Python的FastAPI框架实现的服务，用于统计客户端的心跳信息，并据此维护在线用户列表以及记录活跃用户数。

功能特性

• 心跳接收：接受来自客户端的心跳包，以更新客户端的状态。
• 在线用户统计：提供API接口来获取当前在线用户的数量。
• 活跃用户统计：提供API接口来获取最近指定天数内活跃的用户数量。
• 请求频率限制：对每个IP地址实施每秒一次的请求频率限制。

安装与运行

1. 请确保已经安装了Python 3.10+。
2. 克隆或下载项目源代码到本地。
3. 在项目根目录下安装所需的依赖库：

   pip install fastapi uvicorn
   

4. 运行服务：

   python main.py
   

   或者使用uvicorn命令直接运行（假设文件名为main.py）：

   uvicorn main:app --reload --host 0.0.0.0 --port 8001
   

API 文档

http://127.0.0.1:8001/docs

心跳接收

• URL: /heartbeat
• 方法: POST
• 描述: 接收客户端发送的心跳信号，并更新客户端为在线状态。
• 响应:
  • 200 OK: 返回JSON格式的信息确认收到心跳。
  • 429 Too Many Requests: 如果客户端在1秒内发送了多个请求。

获取在线用户数量

• URL: /online_clients
• 方法: POST
• 描述: 返回当前在线的客户端数量。
• 响应:
  • 200 OK: 返回包含在线用户数量的JSON对象。
  • 429 Too Many Requests: 请求过于频繁。

获取活跃用户数量

• URL: /total_users
• 方法: GET 或 POST
• 参数:
  • days (可选, 默认值为7): 指定要查询的天数。
• 描述: 返回最近几天内有活动记录的用户数量。
• 响应:
  • 200 OK: 返回包含活跃用户数量和查询天数的JSON对象。
  • 429 Too Many Requests: 请求过于频繁。

数据存储

所有客户端的心跳时间戳将被持久化到一个JSON文件中，该文件位于服务启动时所在的目录下的users_data.json。每次接收到新的心跳信号时，都会更新此文件。

注意事项

• 本服务仅用于演示目的，实际生产环境中可能需要考虑更健壮的数据存储解决方案、安全性增强措施等。
• 为了保护服务器免受滥用，已实施了基本的请求频率限制。根据实际需求，可以调整这个限制。
• 服务默认监听在8001端口上，可以通过修改uvicorn.run函数中的port参数来更改。

希望这份文档能对你有所帮助！如果有任何问题或需要进一步的帮助，请随时告诉我。

源码 main.py

from fastapi import FastAPI, Request, Depends, HTTPException
from collections import defaultdict
from datetime import datetime, timedelta
import asyncio
import json
import osapp = FastAPI()# 存储客户端的心跳数据
clients_last_heartbeat = defaultdict(datetime)# 每个IP请求时间间隔限制为1秒
last_request_time = defaultdict(datetime)# 在线客户端统计
online_clients = set()# 心跳超时时间设置为1分钟
HEARTBEAT_TIMEOUT = timedelta(minutes=10)# 用户数据文件路径
USER_DATA_FILE = "users_data.json"# 加载用户数据
def load_user_data():if os.path.exists(USER_DATA_FILE):with open(USER_DATA_FILE, "r") as f:return json.load(f)return {}# 保存用户数据
def save_user_data(data):with open(USER_DATA_FILE, "w") as f:json.dump(data, f)# 初始化用户数据
all_users = load_user_data()@app.on_event("startup")
async def startup_event():# 启动后台任务，每1分钟检查一次在线设备asyncio.create_task(remove_offline_clients())async def remove_offline_clients():"""定时任务：移除超过心跳超时时间未发送心跳的客户端"""while True:await asyncio.sleep(HEARTBEAT_TIMEOUT.total_seconds())now = datetime.utcnow()# 找出超过超时时间未发送心跳的设备，并将其从在线列表中移除offline_clients = {ip for ip, last_heartbeat in clients_last_heartbeat.items()if now - last_heartbeat > HEARTBEAT_TIMEOUT}# 从在线设备中移除离线的客户端for client in offline_clients:online_clients.discard(client)del clients_last_heartbeat[client]  # 删除心跳记录print(f"清除离线客户端, 当前在线客户端数量: {len(online_clients)}")# 请求频率限制，1秒内只能请求一次
def request_limit(request: Request):client_ip = request.client.hostnow = datetime.utcnow()if client_ip in last_request_time and (now - last_request_time[client_ip]).total_seconds() < 1:raise HTTPException(status_code=429, detail="Too Many Requests")last_request_time[client_ip] = now@app.post("/heartbeat")
async def receive_heartbeat(request: Request, limit: None = Depends(request_limit)):"""接受客户端的心跳包"""client_ip = request.client.hostnow = datetime.utcnow()# 更新心跳时间并将客户端标记为在线clients_last_heartbeat[client_ip] = nowonline_clients.add(client_ip)# 更新所有用户数据并保存到文件all_users[client_ip] = now.isoformat()save_user_data(all_users)return {"message": "Heartbeat received", "ip": client_ip}@app.get("/online_clients")
async def get_online_clients(request: Request, limit: None = Depends(request_limit)):"""获取当前在线客户端数量"""return {"online_clients_count": len(online_clients)}@app.post("/online_clients")
async def get_online_clients2(request: Request, limit: None = Depends(request_limit)):"""获取当前在线客户端数量"""return {"online_clients_count": len(online_clients)}@app.get("/total_users")
async def get_total_users(days: int = 7, request: Request = None, limit: None = Depends(request_limit)):"""获取最近n天活跃的用户数"""now = datetime.utcnow()cutoff_date = now - timedelta(days=days)# 筛选最近n天活跃的用户recent_users_count = sum(1 for last_seen in all_users.values()if datetime.fromisoformat(last_seen) >= cutoff_date)return {"recent_users_count": recent_users_count, "days": days}if __name__ == '__main__':import uvicornuvicorn.run(app, host="0.0.0.0", port=8001)