Django中實現(xiàn)一個高性能計數(shù)器(Counter)實例
計數(shù)器(Counter)是一個非常常用的功能組件，這篇blog以未讀消息數(shù)為例，介紹了在 Django中實現(xiàn)一個高性能計數(shù)器的基本要點。
故事的開始：.count()
假設你有一個Notification Model類，保存的主要是所有的站內通知：
class Notification(models.Model):
"""一個簡化過的Notification類，擁有三個字段：
- `user_id`: 消息所有人的用戶ID
- `has_readed`: 表示消息是否已讀
"""
user_id = models.IntegerField(db_index=True)
has_readed = models.BooleanField(default=False)
理所當然的，剛開始你會通過這樣的查詢來獲取某個用戶的未讀消息數(shù)：
# 獲取ID為3074的用戶的未讀消息數(shù)
Notification.objects.filter(user_id=3074, has_readed=False).count()
當你的Notification表比較小的時候，這樣的方式沒有任何的問題，但是慢慢的，隨著業(yè)務量 的擴大。消息表里面有了 上億條數(shù)據(jù) 。很多懶惰的用戶的未讀消息數(shù)都到了上千條。
這時候，你就需要實現(xiàn)一個計數(shù)器，讓這個計數(shù)器來統(tǒng)計每個用戶的未讀消息數(shù)，這樣 比起之前的 count() ，我們只需要執(zhí)行一條簡單的主鍵查詢（或者更優(yōu)）就可以拿到實時的未讀消息數(shù)了。
更優(yōu)的方案：建立計數(shù)器
首先，讓我們得建立一個新表來存儲每個用戶的未讀消息數(shù)。
class UserNotificationsCount(models.Model):
"""這個Model保存著每一個用戶的未讀消息數(shù)目"""
user_id = models.IntegerField(primary_key=True)
unread_count = models.IntegerField(default=0)
def __str__(self):
return '<UserNotificationsCount %s: %s>' % (self.user_id, self.unread_count)
我們?yōu)槊恳粋€注冊用戶提供了一條對應的 UserNotificationsCount 記錄來保存他的未讀消息數(shù)。 每次獲取他的未讀消息數(shù)的時候，只需要 UserNotificationsCount.objects.get(pk=user_id).unread_count 就可以了。
接下來，問題的重點來了，我們如何知道什么時候應該更新我們的計數(shù)器？Django在這方面提供了什么捷徑嗎？
挑戰(zhàn)：實時更新你的計數(shù)器
為了讓我們的計數(shù)器正常的工作，我們必須實時的更新它，這包括：
1.當有新的未讀消息過來的時候，為計數(shù)器 +1
2.當消息被異常刪除時，如果關聯(lián)的消息為未讀，為計數(shù)器 -1
3.當閱讀完一個新消息的時候，為計數(shù)器 -1
讓我們一個一個來解決這些情況。
在拋出解決方案之前，我們需要先介紹Django中的一個功能： Signals ，Signals是django提供的一個事件通知機制，它可以讓你在監(jiān)聽某些自定義或者 預設的事件，當這些事件發(fā)生的時候，調用實現(xiàn)定義好的方法。
比如 django.db.models.signals.pre_save & django.db.models.signals.post_save 表示的是 某個Model調用save方法之前和之后會觸發(fā)的事件，它和Database提供的觸發(fā)器在功能上有一點相似。
關于Signals更多的介紹可以參考官方文檔，下面讓我們來看看Signals能給我們的計數(shù)器帶來什么好處。
1. 當有新的消息過來的時候，為計數(shù)器 +1
這個情況應該是最好處理的，使用Django的Signals，只需要短短幾行代碼，我們便可以實現(xiàn)這種 情況下的計數(shù)器更新：
from django.db.models.signals import post_save, post_delete
def incr_notifications_counter(sender, instance, created, **kwargs):
# 只有當這個instance是新創(chuàng)建，而且has_readed是默認的false才更新
if not (created and not instance.has_readed):
return
# 調用 update_unread_count 方法來更新計數(shù)器 +1
NotificationController(instance.user_id).update_unread_count(1)
# 監(jiān)聽Notification Model的post_save信號
post_save.connect(incr_notifications_counter, sender=Notification)
這樣，每當你使用 Notification.create 或者 .save() 之類的方法創(chuàng)建新通知 時，我們的 NotificationController 便會得到通知，為計數(shù)器 +1。
但是請注意，因為我們的計數(shù)器是基于Django的signals，如果你的代碼里面有地方 在使用原始sql，沒有通過Django ORM方法來添加新通知的話，我們的計數(shù)器是不會得到 通知的，所以，最好規(guī)范所有的新通知建立方式，比如使用同一個API。
2. 當消息被異常刪除時，如果關聯(lián)的消息為未讀，為計數(shù)器 -1
有了第一個的經驗，這種情況處理起來也比較簡單，只需要監(jiān)控Notification的post_delete 信號就可以了，下面是一段實例代碼：
def decr_notifications_counter(sender, instance, **kwargs):
# 當刪除的消息還沒有被讀過時，計數(shù)器 -1
if not instance.has_readed:
NotificationController(instance.user_id).update_unread_count(-1)
post_delete.connect(decr_notifications_counter, sender=Notification)
至此，Notification的刪除事件也能正常的更新我們的計數(shù)器了。
3. 當閱讀一個新消息的時候，為計數(shù)器 -1
接下來，當用戶閱讀某條未讀消息的時候，我們也需要更新我們的未讀消息計數(shù)器。 你可能會說，這有什么難的？我只要在我的閱讀消息的方法里面，手動更新我的計數(shù)器不就好了？
比如這樣：
class NotificationController(object):
... ...
def mark_as_readed(self, notification_id):
notification = Notification.objects.get(pk=notification_id)
# 沒有必要重復標記一個已經讀過的通知
if notication.has_readed:
return
notification.has_readed = True
notification.save()
# 在這里更新我們的計數(shù)器，嗯，我感覺好極了
self.update_unread_count(-1)
通過一些簡單的測試，你可以會覺得你的計數(shù)器工作的非常好，但是，這樣的實現(xiàn)方式有一個 非常致命的問題， 這個方式沒有辦法正常處理并發(fā)的請求 。
打一個比方，你擁有一個id為100的未讀消息對象，這個時候同時有了兩個請求過來，都要標記這個通知為已讀：
# 因為兩個并發(fā)的請求，假設這兩個方法幾乎同時被調用
NotificationController(user_id).mark_as_readed(100)
NotificationController(user_id).mark_as_readed(100)
顯而易見的，這兩次方法都會成功的標記這條通知為已讀，因為在并發(fā)的情況下， if notification.has_readed 這樣的檢查無法正常工作，所以我們的計數(shù)器將會被錯誤的 -1 兩次 ，但其實我們只讀了一條請求。
那么，這樣的問題應該怎么解決呢？
基本上，解決并發(fā)請求產生的數(shù)據(jù)沖突只有一個辦法： 加鎖 ，介紹兩種比較簡單的解決方案：
使用 select for update 數(shù)據(jù)庫查詢
select ... for update 是數(shù)據(jù)庫層面上專門用來解決并發(fā)取數(shù)據(jù)后再修改的場景的，主流的關系數(shù)據(jù)庫 比如mysql、postgresql都支持這個功能， 新版的Django ORM甚至直接提供了這個功能的shortcut 。 關于它的更多介紹，你可以搜索你使用的數(shù)據(jù)庫的介紹文檔。
使用 select for update 后，我們的代碼可能會變成這樣：
from django.db import transaction
class NotificationController(object):
... ...
def mark_as_readed(self, notification_id):
# 手動讓select for update和update語句發(fā)生在一個完整的事務里面
with transaction.commit_on_success():
# 使用select_for_update來保證并發(fā)請求同時只有一個請求在處理，其他的請求
# 等待鎖釋放
notification = Notification.objects.select_for_update().get(pk=notification_id)
# 沒有必要重復標記一個已經讀過的通知
if notication.has_readed:
return
notification.has_readed = True
notification.save()
# 在這里更新我們的計數(shù)器，嗯，我感覺好極了
self.update_unread_count(-1)
除了使用``select for update``這樣的功能，還有一個比較簡單的辦法來解決這個問題。
使用update來實現(xiàn)原子性修改
其實，更簡單的辦法，只要把我們的數(shù)據(jù)庫改成單條的update就可以解決并發(fā)情況下的問題了：
def mark_as_readed(self, notification_id):
affected_rows = Notification.objects.filter(pk=notification_id, has_readed=False)\
.update(has_readed=True)
# affected_rows將會返回update語句修改的條目數(shù)
self.update_unread_count(affected_rows)
這樣，并發(fā)的標記已讀操作也可以正確的影響到我們的計數(shù)器了。
高性能？
我們在之前介紹了如何實現(xiàn)一個能夠正確更新的未讀消息計數(shù)器，我們可能會直接使用UPDATE 語句來修改我們的計數(shù)器，就像這樣：
from django.db.models import F
def update_unread_count(self, count)
# 使用Update語句來更新我們的計數(shù)器
UserNotificationsCount.objects.filter(pk=self.user_id)\
.update(unread_count=F('unread_count') + count)
但是在生產環(huán)境中，這樣的處理方式很有可能造成嚴重的性能問題，因為如果我們的計數(shù)器在頻繁 更新的話，海量的Update會給數(shù)據(jù)庫造成不小的壓力。所以為了實現(xiàn)一個高性能的計數(shù)器，我們 需要把改動暫存起來，然后批量寫入到數(shù)據(jù)庫。
使用 redis 的 sorted set ，我們可以非常輕松的做到這一點。
使用sorted set來緩存計數(shù)器改動
redis是一個非常好用的內存數(shù)據(jù)庫，其中的sorted set是它提供的一種數(shù)據(jù)類型：有序集合， 使用它，我們可以非常簡單的緩存所有的計數(shù)器改動，然后批量回寫到數(shù)據(jù)庫。
RK_NOTIFICATIONS_COUNTER = 'ss_pending_counter_changes'
def update_unread_count(self, count):
"""修改過的update_unread_count方法"""
redisdb.zincrby(RK_NOTIFICATIONS_COUNTER, str(self.user_id), count)
# 同時我們也需要修改獲取用戶未讀消息數(shù)方法，使其獲取redis中那些沒有被回寫
# 到數(shù)據(jù)庫的緩沖區(qū)數(shù)據(jù)。在這里代碼就省略了
通過以上的代碼，我們把計數(shù)器的更新緩沖在了redis里面，我們還需要一個腳本來把這個緩沖區(qū) 里面的數(shù)據(jù)定時回寫到數(shù)據(jù)庫中。
通過自定義django的command，我們可以非常輕松的做到這一點：
# File: management/commands/notification_update_counter.py
# -*- coding: utf-8 -*-
from django.core.management.base import BaseCommand
from django.db.models import F
# Fix import prob
from notification.models import UserNotificationsCount
from notification.utils import RK_NOTIFICATIONS_COUNTER
from base_redis import redisdb
import logging
logger = logging.getLogger('stdout')
class Command(BaseCommand):
help = 'Update UserNotificationsCounter objects, Write changes from redis to database'
def handle(self, *args, **options):
# 首先，通過 zrange 命令來獲取緩沖區(qū)所有修改過的用戶ID
for user_id in redisdb.zrange(RK_NOTIFICATIONS_COUNTER, 0, -1):
# 這里值得注意，為了保證操作的原子性，我們使用了redisdb的pipeline
pipe = redisdb.pipeline()
pipe.zscore(RK_NOTIFICATIONS_COUNTER, user_id)
pipe.zrem(RK_NOTIFICATIONS_COUNTER, user_id)
count, _ = pipe.execute()
count = int(count)
if not count:
continue
logger.info('Updating unread count user %s: count %s' % (user_id, count))
UserNotificationsCount.objects.filter(pk=obj.pk)\
.update(unread_count=F('unread_count') + count)
之后，通過 python manage.py notification_update_counter 這樣的命令就可以把緩沖區(qū) 里面的改動批量回寫到數(shù)據(jù)庫了。我們還可以把這個命令配置到crontab中來定義執(zhí)行。
總結
文章到了這里，一個簡單的“高性能”未讀消息計數(shù)器算是實現(xiàn)完了。說了這么多，其實主要的知識點就是這么些：
1.使用Django的signals來獲取Model的新建/刪除操作更新
2.使用數(shù)據(jù)庫的select for update來正確處理并發(fā)的數(shù)據(jù)庫操作
3.使用redis的sorted set來緩存計數(shù)器的修改操作
希望能對您有所幫助。 :)
相關文章
使用python和opencv的mask實現(xiàn)摳圖疊加
這篇文章主要介紹了使用python和opencv的mask實現(xiàn)摳圖疊加操作，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2021-04-04
這篇文章主要介紹了python動態(tài)網(wǎng)頁批量爬取的方法，主要針對四六級成績批量爬取，感興趣的小伙伴們可以參考一下2016-02-02
本篇文章通過代碼實例給大家分享了python實現(xiàn)裝飾器、描述符模擬代碼的過程，對戲需要的讀者可以參考下。2018-02-02
這篇文章主要介紹了Python實現(xiàn)的KMeans聚類算法,結合實例形式較為詳細的分析了KMeans聚類算法概念、原理、定義及使用相關操作技巧,需要的朋友可以參考下2018-12-12
這篇文章主要介紹了python自然語言編碼轉換模塊codecs介紹,codecs專門用作編碼轉換,通過它的接口是可以擴展到其他關于代碼方面的轉換,需要的朋友可以參考下2015-04-04
這篇文章主要介紹了簡單的抓取淘寶圖片的Python爬蟲,方法比較簡單，推薦給小伙伴們參考下。2014-12-12
這篇文章主要介紹了Python 責任鏈設計模式的相關資料，幫助大家更好的理解和學習python 設計模式，感興趣的朋友可以了解下2020-09-09
python中pygame針對游戲窗口的顯示方法實例分析(附源碼)
這篇文章主要介紹了python中pygame針對游戲窗口的顯示方法,以完整實例形式較為詳細的分析了pygame響應鍵盤按鍵改變窗口顯示效果的相關實現(xiàn)技巧,需要的朋友可以參考下2015-11-11
使用不同的Python版本，如果我們把需要的不同版本的Python都下載到服務器上，管理起來會非常困難，多版本并存又容易互相干擾,接下來就來介紹一個Python環(huán)境管理工具：pyenv，就可輕松的在多個版本的Python之間自由切換,需要的朋友可以參考下2021-09-09
導入tensorflow:ImportError: libcublas.so.9.0 報錯
這篇文章主要介紹了導入tensorflow:ImportError: libcublas.so.9.0 報錯，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2020-01-01