首頁  >  文章  >  後端開發  >  Python的Django框架中訊息通知的計數器

Python的Django框架中訊息通知的計數器

高洛峰
高洛峰原創
2017-03-03 13:14:111169瀏覽

故事的開始:.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)

理所當然的,剛開始你會透過這樣的查詢來取得某個使用者的未讀訊息數:

# 获取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 &#39;<UserNotificationsCount %s: %s>&#39; % (self.user_id, self.unread_count)

我們為每個註冊使用者提供了一個對應的 UserNotificationsCount 記錄來保存他的未讀訊息數。 每次取得他的未讀取訊息數的時候,只需要 UserNotificationsCount.objects.get(pk=user_id).unread_count 就可以了。

接下來,問題的重點來了,我們如何知道什麼時候應該更新我們的計數器? Django在這方面提供了什麼捷徑嗎?

挑戰:即時更新你的計數器

為了讓我們的計數器正常的工作,我們必須即時的更新它,這包括:

  • 當有新的未讀訊息過來的時候,為計數器+1

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

  • 當閱讀完一個新訊息​​的時候,為計數器-1

讓我們一個一個來解決這些情況。

在拋出解決方案之前,我們需要先介紹Django中的一個功能: Signals ,Signals是django提供的一個事件通知機制,它可以讓你在監聽某些自訂或預設的事件,當這些事件發生的時候,呼叫實作定義好的方法。

例如django.db.models.signals.pre_save & django.db.models.signals.post_save 表示的是某個Model呼叫save方法之前和之後會觸發的事件,它和Database提供的觸發器在功能上有一點相似。

關於Signals更多的介紹可以參考官方文檔,以下讓我們來看看Signals能為我們的計數器帶來什麼好處。

1. 當有新的訊息過來的時候,為計數器+1

這個情況應該是最好處理的,使用Django的Signals,只需要短短幾行程式碼,我們便可以實現這種情況下的計數器更新:

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。

但請注意,因為我們的計數器是基於Django的signals,如果你的程式碼裡面有地方在使用原始sql,沒有透過Django ORM方法來新增通知的話,我們的計數器是不會得到通知的,所以,最好規範所有的新通知建立方式,例如使用同一個API。

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

有了第一個的經驗,這種情況處理起來也比較簡單,只需要監控Notification的post_delete 訊號就可以了,以下是一段實例程式碼:

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)


至此,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(&#39;unread_count&#39;) + count)

但是在生产环境中,这样的处理方式很有可能造成严重的性能问题,因为如果我们的计数器在频繁 更新的话,海量的Update会给数据库造成不小的压力。所以为了实现一个高性能的计数器,我们 需要把改动暂存起来,然后批量写入到数据库。

使用 redis 的 sorted set ,我们可以非常轻松的做到这一点。

使用sorted set来缓存计数器改动

redis是一个非常好用的内存数据库,其中的sorted set是它提供的一种数据类型:有序集合, 使用它,我们可以非常简单的缓存所有的计数器改动,然后批量回写到数据库。

RK_NOTIFICATIONS_COUNTER = &#39;ss_pending_counter_changes&#39;

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(&#39;stdout&#39;)


class Command(BaseCommand):
  help = &#39;Update UserNotificationsCounter objects, Write changes from redis to database&#39;

  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(&#39;Updating unread count user %s: count %s&#39; % (user_id, count))
      UserNotificationsCount.objects.filter(pk=obj.pk)\
                     .update(unread_count=F(&#39;unread_count&#39;) + count)

之后,通过 python manage.py notification_update_counter 这样的命令就可以把缓冲区 里面的改动批量回写到数据库了。我们还可以把这个命令配置到crontab中来定义执行。

总结
文章到了这里,一个简单的“高性能”未读消息计数器算是实现完了。说了这么多,其实主要的知识点就是这么些:

使用Django的signals来获取Model的新建/删除操作更新
使用数据库的select for update来正确处理并发的数据库操作
使用redis的sorted set来缓存计数器的修改操作
希望能对您有所帮助。 :)

更多Python的Django框架中消息通知的计数器相关文章请关注PHP中文网!

陳述:
本文內容由網友自願投稿,版權歸原作者所有。本站不承擔相應的法律責任。如發現涉嫌抄襲或侵權的內容,請聯絡admin@php.cn