簡易在線用戶統(tǒng)計服務(wù)

概述

這是一個基于Python的FastAPI框架實現(xiàn)的服務(wù)，用于統(tǒng)計客戶端的心跳信息，并據(jù)此維護在線用戶列表以及記錄活躍用戶數(shù)。

功能特性

• 心跳接收：接受來自客戶端的心跳包，以更新客戶端的狀態(tài)。
• 在線用戶統(tǒng)計：提供API接口來獲取當(dāng)前在線用戶的數(shù)量。
• 活躍用戶統(tǒng)計：提供API接口來獲取最近指定天數(shù)內(nèi)活躍的用戶數(shù)量。
• 請求頻率限制：對每個IP地址實施每秒一次的請求頻率限制。

安裝與運行

1. 請確保已經(jīng)安裝了Python 3.10+。
2. 克隆或下載項目源代碼到本地。
3. 在項目根目錄下安裝所需的依賴庫：

   pip install fastapi uvicorn
   

4. 運行服務(wù)：

   python main.py
   

   或者使用uvicorn命令直接運行（假設(shè)文件名為main.py）：

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

API 文檔

http://127.0.0.1:8001/docs

心跳接收

• URL: /heartbeat
• 方法: POST
• 描述: 接收客戶端發(fā)送的心跳信號，并更新客戶端為在線狀態(tài)。
• 響應(yīng):
  • 200 OK: 返回JSON格式的信息確認收到心跳。
  • 429 Too Many Requests: 如果客戶端在1秒內(nèi)發(fā)送了多個請求。

獲取在線用戶數(shù)量

• URL: /online_clients
• 方法: POST
• 描述: 返回當(dāng)前在線的客戶端數(shù)量。
• 響應(yīng):
  • 200 OK: 返回包含在線用戶數(shù)量的JSON對象。
  • 429 Too Many Requests: 請求過于頻繁。

獲取活躍用戶數(shù)量

• URL: /total_users
• 方法: GET 或 POST
• 參數(shù):
  • days (可選, 默認值為7): 指定要查詢的天數(shù)。
• 描述: 返回最近幾天內(nèi)有活動記錄的用戶數(shù)量。
• 響應(yīng):
  • 200 OK: 返回包含活躍用戶數(shù)量和查詢天數(shù)的JSON對象。
  • 429 Too Many Requests: 請求過于頻繁。

數(shù)據(jù)存儲

所有客戶端的心跳時間戳將被持久化到一個JSON文件中，該文件位于服務(wù)啟動時所在的目錄下的users_data.json。每次接收到新的心跳信號時，都會更新此文件。

注意事項

• 本服務(wù)僅用于演示目的，實際生產(chǎn)環(huán)境中可能需要考慮更健壯的數(shù)據(jù)存儲解決方案、安全性增強措施等。
• 為了保護服務(wù)器免受濫用，已實施了基本的請求頻率限制。根據(jù)實際需求，可以調(diào)整這個限制。
• 服務(wù)默認監(jiān)聽在8001端口上，可以通過修改uvicorn.run函數(shù)中的port參數(shù)來更改。

希望這份文檔能對你有所幫助！如果有任何問題或需要進一步的幫助，請隨時告訴我。

源碼 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()# 存儲客戶端的心跳數(shù)據(jù)
clients_last_heartbeat = defaultdict(datetime)# 每個IP請求時間間隔限制為1秒
last_request_time = defaultdict(datetime)# 在線客戶端統(tǒng)計
online_clients = set()# 心跳超時時間設(shè)置為1分鐘
HEARTBEAT_TIMEOUT = timedelta(minutes=10)# 用戶數(shù)據(jù)文件路徑
USER_DATA_FILE = "users_data.json"# 加載用戶數(shù)據(jù)
def load_user_data():if os.path.exists(USER_DATA_FILE):with open(USER_DATA_FILE, "r") as f:return json.load(f)return {}# 保存用戶數(shù)據(jù)
def save_user_data(data):with open(USER_DATA_FILE, "w") as f:json.dump(data, f)# 初始化用戶數(shù)據(jù)
all_users = load_user_data()@app.on_event("startup")
async def startup_event():# 啟動后臺任務(wù)，每1分鐘檢查一次在線設(shè)備asyncio.create_task(remove_offline_clients())async def remove_offline_clients():"""定時任務(wù)：移除超過心跳超時時間未發(fā)送心跳的客戶端"""while True:await asyncio.sleep(HEARTBEAT_TIMEOUT.total_seconds())now = datetime.utcnow()# 找出超過超時時間未發(fā)送心跳的設(shè)備，并將其從在線列表中移除offline_clients = {ip for ip, last_heartbeat in clients_last_heartbeat.items()if now - last_heartbeat > HEARTBEAT_TIMEOUT}# 從在線設(shè)備中移除離線的客戶端for client in offline_clients:online_clients.discard(client)del clients_last_heartbeat[client]  # 刪除心跳記錄print(f"清除離線客戶端, 當(dāng)前在線客戶端數(shù)量: {len(online_clients)}")# 請求頻率限制，1秒內(nèi)只能請求一次
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()# 更新心跳時間并將客戶端標(biāo)記為在線clients_last_heartbeat[client_ip] = nowonline_clients.add(client_ip)# 更新所有用戶數(shù)據(jù)并保存到文件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)):"""獲取當(dāng)前在線客戶端數(shù)量"""return {"online_clients_count": len(online_clients)}@app.post("/online_clients")
async def get_online_clients2(request: Request, limit: None = Depends(request_limit)):"""獲取當(dāng)前在線客戶端數(shù)量"""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天活躍的用戶數(shù)"""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)