利用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中來定義執行。