利用django信號實現計數功能


本文主要知識點:

  1.使用Django的signals來獲取Model的新建/刪除操作更新
  2.使用數據庫的select for update來正確處理並發的數據庫操作
  3.使用redis的sorted set來緩存計數器的修改操作

一,起始

  我們要實現消息未讀記錄,常規做法:

class Notification(models.Model):
    """一個簡化過的Notification類,擁有三個字段:
    - `user_id`: 消息所有人的用戶ID
    - `has_readed`: 表示消息是否已讀
    """

    user_id = models.IntegerField(db_index=True)
    has_readed = models.BooleanField(default=False)

  定義一個類,當用戶有一條消息未讀,就生成一條記錄。你可以通過以下方式獲取指定用戶未讀消息數目:

# 獲取ID為3074的用戶的未讀消息數
Notification.objects.filter(user_id=3074, has_readed=False).count()

  當你的Notification表比較小的時候,這樣的方式沒有任何問題。但是隨着業務的增大,表中的數據可能有上億條記錄。那么隨之用戶的未讀消息數,也會有所提高。

  這時候,你就需要實現一個計數器,讓這個計數器來統計每個用戶的未讀消息數,這樣 比起之前的 count() ,我們只需要執行一條簡單的主鍵查詢(或者更優)就可以拿到實時的未讀消息數了。

二,優化

  首先 ,我們得建立一個新表來存儲每個用戶的未讀消息數。  

class UserNotificationsCount(models.Model):
    """這個Model保存着每一個用戶的未讀消息數目"""
    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)

  我們為每個注冊的用戶提供一條對應的記錄,來保存用戶的未讀消息數。

  那么重點就是,我們如何知道什么時候應該更新unread_count呢?

  這就要說到Django給我提供的強大的信號機制,從而方便我們實時檢測計數器。

  為了實時更新我們的計數器,我們必須做到以下幾點:

    1.當有新的未讀消息過來的時候,計數器+1

    2.當消息被異常刪除時,如果關聯的消息未讀,為計數器-1

    3.當閱讀完一個新消息的時候,計數器-1

   django.db.models.signals.pre_save & django.db.models.signals.post_save 表示的是 某個Model調用save方法之前和之后會觸發的事件

   現在,我們利用Django提供的兩個信號機制來實現我們的實時更新計數器。

  1.當有新的未讀消息過來的時候,計數器+1

from django.db.models.signals import post_save,post_delete

def incr_notifications_counter(sender,instance,created,**kwargs):

     # 只有當這個instance是新創建,而且has_readed是默認的false才更新
    if not (created and not instance.has_readed):
        return

    # 調用 update_unread_count 方法來更新計數器 +1
    NotificationController(instance.user_id).update_unread_count(1)

# 監聽Notification Model的post_save信號
post_save.connect(incr_notifications_counter,sender=Notification)

  這樣,每當你使用Notification.create或者.save()之類的方法創建新通知時,我們的 NotificationController 便會得到通知,為計數器 +1。

  2.當消息被異常刪除時,如果關聯的消息為未讀,為計數器 -1

def decr_notifications_counter(sender, instance, **kwargs):
    # 當刪除的消息還沒有被讀過時,計數器 -1
    if not instance.has_readed:
        NotificationController(instance.user_id).update_unread_count(-1)
post_delete.connect(decr_notifications_counter, sender=Notification)

  3. 當閱讀一個新消息的時候,為計數器 -1

  接下來,當用戶閱讀某條未讀消息的時候,我們也需要更新我們的未讀消息計數器。 你可能會說,這有什么難的?我只要在我的閱讀消息的方法里面,手動更新我的計數器不就好了?

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()
        # 在這里更新我們的計數器,嗯,我感覺好極了
        self.update_unread_count(-1)

  但是,這樣的實現方式有一個 非常致命的問題, 這個方式沒有辦法正常處理並發的請求。

  例如,你擁有一個id為100的未讀消息對象,這個時候同時有了兩個請求過來,都要標記這個通知為已讀

# 因為兩個並發的請求,假設這兩個方法幾乎同時被調用
NotificationController(user_id).mark_as_readed(100)
NotificationController(user_id).mark_as_readed(100)

  顯而易見的,這兩次方法都會成功的標記這條通知為已讀,因為在並發的情況下, if notification.has_readed 這樣的檢查無法正常工作,所以我們的計數器將會被錯誤的 -1 兩次 ,但其實我們只讀了一條請求。

  那么,這樣的問題應該怎么解決呢?

  基本上,解決並發請求產生的數據沖突只有一個辦法: 加鎖 ,介紹兩種比較簡單的解決方案:

  使用 select for update 數據庫查詢

  select ... for update 是數據庫層面上專門用來解決並發取數據后再修改的場景的,主流的關系數據庫 比如mysql、postgresql都支持這個功能, 新版的Django ORM甚至直接提供了這個功能的shortcut 。 關於它的更多介紹,你可以搜索你使用的數據庫的介紹文檔。

  使用 select for update 后,我們的代碼可能會變成這樣:

from django.db import transaction
class NotificationController(object):
  
def mark_as_readed(self, notification_id): # 手動讓select for update和update語句發生在一個完整的事務里面 with transaction.commit_on_success(): # 使用select_for_update來保證並發請求同時只有一個請求在處理,其他的請求 # 等待鎖釋放 notification = Notification.objects.select_for_update().get(pk=notification_id) # 沒有必要重復標記一個已經讀過的通知 if notication.has_readed: return notification.has_readed = True notification.save() # 在這里更新我們的計數器,嗯,我感覺好極了 self.update_unread_count(-1)

  除了使用``select for update``這樣的功能,還有一個比較簡單的辦法來解決這個問題。

  使用update來實現原子性修改

  其實,更簡單的辦法,只要把我們的數據庫改成單條的update就可以解決並發情況下的問題了:

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語句修改的條目數
        self.update_unread_count(affected_rows)

  這樣,並發的標記已讀操作也可以正確的影響到我們的計數器了。

三,高性能

  我們可能會直接使用UPDATE 語句來修改我們的計數器,就像這樣:

from django.db.models import F
def update_unread_count(self, count)
    # 使用Update語句來更新我們的計數器
    UserNotificationsCount.objects.filter(pk=self.user_id).update(unread_count=F('unread_count') + count)

  但是在生產環境中,這樣的處理方式很有可能造成嚴重的性能問題,因為如果我們的計數器在頻繁 更新的話,海量的Update會給數據庫造成不小的壓力。所以為了實現一個高性能的計數器,我們 需要把改動暫存起來,然后批量寫入到數據庫。

  使用 redis 的 sorted set ,我們可以非常輕松的做到這一點。

  使用sorted set來緩存計數器改動

  redis是一個非常好用的內存數據庫,其中的sorted set是它提供的一種數據類型:有序集合, 使用它,我們可以非常簡單的緩存所有的計數器改動,然后批量回寫到數據庫。

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)

# 同時我們也需要修改獲取用戶未讀消息數方法,使其獲取redis中那些沒有被回寫
# 到數據庫的緩沖區數據。在這里代碼就省略了

  通過以上的代碼,我們把計數器的更新緩沖在了redis里面,我們還需要一個腳本來把這個緩沖區 里面的數據定時回寫到數據庫中。

  通過自定義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 命令來獲取緩沖區所有修改過的用戶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 這樣的命令就可以把緩沖區 里面的改動批量回寫到數據庫了。我們還可以把這個命令配置到crontab中來定義執行。

  

 


免責聲明!

本站轉載的文章為個人學習借鑒使用,本站對版權不負任何法律責任。如果侵犯了您的隱私權益,請聯系本站郵箱yoyou2525@163.com刪除。



 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM