Flask 統計在線人數

服務端完成以後，如果檢驗應用的效果呢，在線人數/客戶端是一個不錯的指標。但是客戶端的連接通常是短連接「請求建立一次連接，請求完成連接即斷開」，基於這種情況服務端需要在每次的客戶端請求時記錄當前的時間，以此來間接實現在線人數/客戶端的統計「比如：5 分鐘內過連接的客戶端認為處於在線狀態」。

在每次請求中記錄下用戶/客戶端的 ID 及當前的時間，可以通過以下方式：

  1. 使用字典直接存儲在內存中。
  2. 在用戶數據表中存儲最後連接的時間。
  3. 使用 radis 來存儲連接信息。

一個服務端總是會有很多的 API 接口，要統計每個連接的時間，我們總不能在每個 API 接口下都寫一遍統計函數吧「這樣也太不 python 了」， python 的方式應該是在 flask_httpatuh 的 verify_token 和 verify_password 裝飾器中進行處理。

@token_auth.verify_token  def verify_token(token):      g.current_user = None      user = User.verify_auth_token(token, current_app)      if not user or not user.isActive():          return False      g.current_user = user      mark_online(g.current_user)      return True    @basic_auth.verify_password  def verify_password(username, password):      g.current_user = None      user = User.query.filter_by(name = username).first()      if not user or not user.check_password(password) or not user.isActive():          return False      g.current_user = user      mark_online(g.current_user)      return True

使用字典直接存儲在內存中

使用字典來存儲最後連接時間，直接將用戶 id 作為 kye 將時間作為 value 存入字典中，獲取在線人數時，直接遍歷字典比較時間即可。

import time    _inline_user = {}    def make_inline(user):      _inline_user[user.id] = int(time.time())    def get_online():      count = 0;      minutes = int(time.time()) - 5 * 60      print(minutes)      for _, values in _inline_user.items():          if values > minutes:              count = count + 1      return count

在用戶數據表中存儲最後連接的時間

直接在當前數據表中新增一列用來存儲最後連接時間，需要對用戶表進行修改

class User(BaseModel, UserMixin):      __tablename__ = 'user'        ...      lastseen = db.Column(db.DateTime, default=datetime.datetime.utcnow, nullable=True)      ...

然後在每次連接時將時間同步到數據庫即可，在讀取時直接檢索符合條件的用戶即可。

from datetime import datetime, timedelta  from pytz import UTC    def mark_online(user):      user.lastseen = datetime.now(UTC)      user.save()    def get_online():      diff = datetime.now(UTC) - timedelta(5)      return User.query.filter(User.lastseen >= diff).count()

使用數據庫保存，還可以查看指定時間段內的在線人數「比如 3 天內，5 天內」，只需要更改相應的過濾條件。

使用 redis 來存儲連接信息

使用 redis 來存儲信息，需要在你的電腦「服務器」上已安裝 redis。redis 相關內容請看 https://redis.io/。

Flask 中使用 redis 有方便的第三方庫 flask_redis。可以直接通過 pip 命令進行安裝。

pip install flask-redis

確保 redis 已經在本機正常啟動，並在配置文件中配置 URL：REDIS_URL="redis://localhost:6379"。

from flask_redis import FlaskRedis    redis_client = FlaskRedis()    def mark_online(user):      user_id = str(user.id).encode('utf-8')      now = int(time.time())      expires = now + (5 * 60) + 10      all_users_key = "online-users/%d" % (now // 60)      user_key = "user-activity/%s" % user_id      p = redis_client.pipeline()      p.sadd(all_users_key, user_id)      p.set(user_key, now)      p.expireat(all_users_key, expires)      p.expireat(user_key, expires)      p.execute()    def get_online():      current = int(time.time()) // 60      minutes = range(5)      users = redis_client.sunion(          ["online-users/%d" % (current - x) for x in minutes]      )        return len([int(u.decode('utf-8')) for u in users])

注意：還需要通過 redisclient.initapp(app) 將 flask 對象傳入 flask_redis 對象。

使用字典和 redis 是直接將輸入在內容中存儲，如果系統掉電或意外重啟將丟失所有的數據，直接在數據中存儲會增加數據庫的壓力，各有利弊看個人喜好。

請幫我點個在看吧！

VirMach 便宜 VPS

QNews

QNews