故事的開始：.count()
假設(shè)你有一個(gè)Notification Model類，保存的主要是所有的站內(nèi)通知：

class Notification(models.Model):
  """一個(gè)簡(jiǎn)化過(guò)的Notification類，擁有三個(gè)字段：

  - `user_id`: 消息所有人的用戶ID
  - `has_readed`: 表示消息是否已讀
  """

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

理所當(dāng)然的，剛開始你會(huì)通過(guò)這樣的查詢來(lái)獲取某個(gè)用戶的未讀消息數(shù)：

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

當(dāng)你的Notification表比較小的時(shí)候，這樣的方式?jīng)]有任何的問題，但是慢慢的，隨著業(yè)務(wù)量 的擴(kuò)大。消息表里面有了 上億條數(shù)據(jù) 。很多懶惰的用戶的未讀消息數(shù)都到了上千條。

這時(shí)候，你就需要實(shí)現(xiàn)一個(gè)計(jì)數(shù)器，讓這個(gè)計(jì)數(shù)器來(lái)統(tǒng)計(jì)每個(gè)用戶的未讀消息數(shù)，這樣 比起之前的 count() ，我們只需要執(zhí)行一條簡(jiǎn)單的主鍵查詢（或者更優(yōu)）就可以拿到實(shí)時(shí)的未讀消息數(shù)了。

更優(yōu)的方案：建立計(jì)數(shù)器
首先，讓我們得建立一個(gè)新表來(lái)存儲(chǔ)每個(gè)用戶的未讀消息數(shù)。

class UserNotificationsCount(models.Model):
  """這個(gè)Model保存著每一個(gè)用戶的未讀消息數(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)槊恳粋€(gè)注冊(cè)用戶提供了一條對(duì)應(yīng)的 UserNotificationsCount 記錄來(lái)保存他的未讀消息數(shù)。 每次獲取他的未讀消息數(shù)的時(shí)候，只需要 UserNotificationsCount.objects.get(pk=user_id).unread_count 就可以了。

接下來(lái)，問題的重點(diǎn)來(lái)了，我們?nèi)绾沃朗裁磿r(shí)候應(yīng)該更新我們的計(jì)數(shù)器？Django在這方面提供了什么捷徑嗎？

挑戰(zhàn)：實(shí)時(shí)更新你的計(jì)數(shù)器

為了讓我們的計(jì)數(shù)器正常的工作，我們必須實(shí)時(shí)的更新它，這包括：

• 當(dāng)有新的未讀消息過(guò)來(lái)的時(shí)候，為計(jì)數(shù)器 +1
• 當(dāng)消息被異常刪除時(shí)，如果關(guān)聯(lián)的消息為未讀，為計(jì)數(shù)器 -1
• 當(dāng)閱讀完一個(gè)新消息的時(shí)候，為計(jì)數(shù)器 -1

讓我們一個(gè)一個(gè)來(lái)解決這些情況。

在拋出解決方案之前，我們需要先介紹Django中的一個(gè)功能： Signals ，Signals是django提供的一個(gè)事件通知機(jī)制，它可以讓你在監(jiān)聽某些自定義或者 預(yù)設(shè)的事件，當(dāng)這些事件發(fā)生的時(shí)候，調(diào)用實(shí)現(xiàn)定義好的方法。

比如 django.db.models.signals.pre_save & django.db.models.signals.post_save 表示的是 某個(gè)Model調(diào)用save方法之前和之后會(huì)觸發(fā)的事件，它和Database提供的觸發(fā)器在功能上有一點(diǎn)相似。

關(guān)于Signals更多的介紹可以參考官方文檔，下面讓我們來(lái)看看Signals能給我們的計(jì)數(shù)器帶來(lái)什么好處。

1. 當(dāng)有新的消息過(guò)來(lái)的時(shí)候，為計(jì)數(shù)器 +1

這個(gè)情況應(yīng)該是最好處理的，使用Django的Signals，只需要短短幾行代碼，我們便可以實(shí)現(xiàn)這種 情況下的計(jì)數(shù)器更新：

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

def incr_notifications_counter(sender, instance, created, **kwargs):
  # 只有當(dāng)這個(gè)instance是新創(chuàng)建，而且has_readed是默認(rèn)的false才更新
  if not (created and not instance.has_readed):
    return

  # 調(diào)用 update_unread_count 方法來(lái)更新計(jì)數(shù)器 +1
  NotificationController(instance.user_id).update_unread_count(1)

# 監(jiān)聽Notification Model的post_save信號(hào)
post_save.connect(incr_notifications_counter, sender=Notification)

這樣，每當(dāng)你使用 Notification.create 或者 .save() 之類的方法創(chuàng)建新通知 時(shí)，我們的 NotificationController 便會(huì)得到通知，為計(jì)數(shù)器 +1。

但是請(qǐng)注意，因?yàn)槲覀兊挠?jì)數(shù)器是基于Django的signals，如果你的代碼里面有地方 在使用原始sql，沒有通過(guò)Django ORM方法來(lái)添加新通知的話，我們的計(jì)數(shù)器是不會(huì)得到 通知的，所以，最好規(guī)范所有的新通知建立方式，比如使用同一個(gè)API。

2. 當(dāng)消息被異常刪除時(shí)，如果關(guān)聯(lián)的消息為未讀，為計(jì)數(shù)器 -1

有了第一個(gè)的經(jīng)驗(yàn)，這種情況處理起來(lái)也比較簡(jiǎn)單，只需要監(jiān)控Notification的post_delete 信號(hào)就可以了，下面是一段實(shí)例代碼：

def decr_notifications_counter(sender, instance, **kwargs):
  # 當(dāng)刪除的消息還沒有被讀過(guò)時(shí)，計(jì)數(shù)器 -1
  if not instance.has_readed:
    NotificationController(instance.user_id).update_unread_count(-1)

post_delete.connect(decr_notifications_counter, sender=Notification)

至此，Notification的刪除事件也能正常的更新我們的計(jì)數(shù)器了。

3. 當(dāng)閱讀一個(gè)新消息的時(shí)候，為計(jì)數(shù)器 -1

接下來(lái)，當(dāng)用戶閱讀某條未讀消息的時(shí)候，我們也需要更新我們的未讀消息計(jì)數(shù)器。 你可能會(huì)說(shuō)，這有什么難的？我只要在我的閱讀消息的方法里面，手動(dòng)更新我的計(jì)數(shù)器不就好了？

比如這樣：

class NotificationController(object):

  ... ...

  def mark_as_readed(self, notification_id):
    notification = Notification.objects.get(pk=notification_id)
    # 沒有必要重復(fù)標(biāo)記一個(gè)已經(jīng)讀過(guò)的通知
    if notication.has_readed:
      return

    notification.has_readed = True
    notification.save()
    # 在這里更新我們的計(jì)數(shù)器，嗯，我感覺好極了
    self.update_unread_count(-1)

通過(guò)一些簡(jiǎn)單的測(cè)試，你可以會(huì)覺得你的計(jì)數(shù)器工作的非常好，但是，這樣的實(shí)現(xiàn)方式有一個(gè) 非常致命的問題， 這個(gè)方式?jīng)]有辦法正常處理并發(fā)的請(qǐng)求 。

打一個(gè)比方，你擁有一個(gè)id為100的未讀消息對(duì)象，這個(gè)時(shí)候同時(shí)有了兩個(gè)請(qǐng)求過(guò)來(lái)，都要標(biāo)記這個(gè)通知為已讀：

# 因?yàn)閮蓚€(gè)并發(fā)的請(qǐng)求，假設(shè)這兩個(gè)方法幾乎同時(shí)被調(diào)用
NotificationController(user_id).mark_as_readed(100)
NotificationController(user_id).mark_as_readed(100)

顯而易見的，這兩次方法都會(huì)成功的標(biāo)記這條通知為已讀，因?yàn)樵诓l(fā)的情況下， if notification.has_readed 這樣的檢查無(wú)法正常工作，所以我們的計(jì)數(shù)器將會(huì)被錯(cuò)誤的 -1 兩次 ，但其實(shí)我們只讀了一條請(qǐng)求。

那么，這樣的問題應(yīng)該怎么解決呢？

基本上，解決并發(fā)請(qǐng)求產(chǎn)生的數(shù)據(jù)沖突只有一個(gè)辦法： 加鎖 ，介紹兩種比較簡(jiǎn)單的解決方案：

使用 select for update 數(shù)據(jù)庫(kù)查詢

select ... for update 是數(shù)據(jù)庫(kù)層面上專門用來(lái)解決并發(fā)取數(shù)據(jù)后再修改的場(chǎng)景的，主流的關(guān)系數(shù)據(jù)庫(kù) 比如mysql、postgresql都支持這個(gè)功能， 新版的Django ORM甚至直接提供了這個(gè)功能的shortcut 。 關(guān)于它的更多介紹，你可以搜索你使用的數(shù)據(jù)庫(kù)的介紹文檔。

使用 select for update 后，我們的代碼可能會(huì)變成這樣：

from django.db import transaction

class NotificationController(object):

  ... ...

  def mark_as_readed(self, notification_id):
    # 手動(dòng)讓select for update和update語(yǔ)句發(fā)生在一個(gè)完整的事務(wù)里面
    with transaction.commit_on_success():
      # 使用select_for_update來(lái)保證并發(fā)請(qǐng)求同時(shí)只有一個(gè)請(qǐng)求在處理，其他的請(qǐng)求
      # 等待鎖釋放
      notification = Notification.objects.select_for_update().get(pk=notification_id)
      # 沒有必要重復(fù)標(biāo)記一個(gè)已經(jīng)讀過(guò)的通知
      if notication.has_readed:
        return

      notification.has_readed = True
      notification.save()
      # 在這里更新我們的計(jì)數(shù)器，嗯，我感覺好極了
      self.update_unread_count(-1)

除了使用``select for update``這樣的功能，還有一個(gè)比較簡(jiǎn)單的辦法來(lái)解決這個(gè)問題。

使用update來(lái)實(shí)現(xiàn)原子性修改

其實(shí)，更簡(jiǎn)單的辦法，只要把我們的數(shù)據(jù)庫(kù)改成單條的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將會(huì)返回update語(yǔ)句修改的條目數(shù)
    self.update_unread_count(affected_rows)

這樣，并發(fā)的標(biāo)記已讀操作也可以正確的影響到我們的計(jì)數(shù)器了。

高性能？
我們?cè)谥敖榻B了如何實(shí)現(xiàn)一個(gè)能夠正確更新的未讀消息計(jì)數(shù)器，我們可能會(huì)直接使用UPDATE 語(yǔ)句來(lái)修改我們的計(jì)數(shù)器，就像這樣：

from django.db.models import F

def update_unread_count(self, count)
  # 使用Update語(yǔ)句來(lái)更新我們的計(jì)數(shù)器
  UserNotificationsCount.objects.filter(pk=self.user_id)\
                 .update(unread_count=F('unread_count') + count)

但是在生產(chǎn)環(huán)境中，這樣的處理方式很有可能造成嚴(yán)重的性能問題，因?yàn)槿绻覀兊挠?jì)數(shù)器在頻繁 更新的話，海量的Update會(huì)給數(shù)據(jù)庫(kù)造成不小的壓力。所以為了實(shí)現(xiàn)一個(gè)高性能的計(jì)數(shù)器，我們 需要把改動(dòng)暫存起來(lái)，然后批量寫入到數(shù)據(jù)庫(kù)。

使用 redis 的 sorted set ，我們可以非常輕松的做到這一點(diǎn)。

使用sorted set來(lái)緩存計(jì)數(shù)器改動(dòng)

redis是一個(gè)非常好用的內(nèi)存數(shù)據(jù)庫(kù)，其中的sorted set是它提供的一種數(shù)據(jù)類型：有序集合， 使用它，我們可以非常簡(jiǎn)單的緩存所有的計(jì)數(shù)器改動(dòng)，然后批量回寫到數(shù)據(jù)庫(kù)。

RK_NOTIFICATIONS_COUNTER = 'ss_pending_counter_changes'

def update_unread_count(self, count):
  """修改過(guò)的update_unread_count方法"""
  redisdb.zincrby(RK_NOTIFICATIONS_COUNTER, str(self.user_id), count)

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

通過(guò)以上的代碼，我們把計(jì)數(shù)器的更新緩沖在了redis里面，我們還需要一個(gè)腳本來(lái)把這個(gè)緩沖區(qū) 里面的數(shù)據(jù)定時(shí)回寫到數(shù)據(jù)庫(kù)中。

通過(guò)自定義django的command，我們可以非常輕松的做到這一點(diǎn)：

# 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):
    # 首先，通過(guò) zrange 命令來(lái)獲取緩沖區(qū)所有修改過(guò)的用戶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)

之后，通過(guò) python manage.py notification_update_counter 這樣的命令就可以把緩沖區(qū) 里面的改動(dòng)批量回寫到數(shù)據(jù)庫(kù)了。我們還可以把這個(gè)命令配置到crontab中來(lái)定義執(zhí)行。

總結(jié)
文章到了這里，一個(gè)簡(jiǎn)單的“高性能”未讀消息計(jì)數(shù)器算是實(shí)現(xiàn)完了。說(shuō)了這么多，其實(shí)主要的知識(shí)點(diǎn)就是這么些：

使用Django的signals來(lái)獲取Model的新建/刪除操作更新
使用數(shù)據(jù)庫(kù)的select for update來(lái)正確處理并發(fā)的數(shù)據(jù)庫(kù)操作
使用redis的sorted set來(lái)緩存計(jì)數(shù)器的修改操作
希望能對(duì)您有所幫助。 :)

最新評(píng)論