Python の Django フレームワークでのメッセージ通知のカウンター実装に関するチュートリアル

物語の始まり: .count()
主にすべてのサイト通知を保存する通知モデル クラスがあるとします。

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()

通知テーブルが比較的小さい場合、この方法で問題はありませんが、ビジネス量が拡大するにつれて徐々に遅くなります。メッセージ テーブルには数億のデータがあります。怠惰なユーザーの多くは、何千もの未読メッセージを抱えています。

現時点では、各ユーザーの未読メッセージの数をカウントするカウンターを実装する必要があります。この方法では、以前の 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)

未読メッセージの数を保存するために、各登録ユーザーに対応する UserNotificationsCount レコードを提供します。 未読メッセージの数を取得するたびに必要なのは、UserNotificationsCount.objects.get(pk=user_id).unread_count だけです。

次に、ここが問題の核心です。カウンタをいつ更新する必要があるかをどのようにして知るのでしょうか? Django はこの点に関してショートカットを提供していますか?

チャレンジ: カウンターをリアルタイムで更新します

カウンターが適切に機能するには、以下を含むカウンターをリアルタイムで更新する必要があります。

• 新しい未読メッセージが到着すると、カウンターに 1 を追加します
• メッセージが異常に削除された場合、関連するメッセージが未読の場合、カウンターは -1 になります
• 新しいメッセージが読まれると、カウンターは -1 になります

これらの状況に一つずつ対処しましょう。

ソリューションを放棄する前に、Django に関数を導入する必要があります。Signals は、Django が提供するイベント通知メカニズムです。イベントが発生したときに、実装定義のメソッドが呼び出されます。

たとえば、django.db.models.signals.pre_save と django.db.models.signals.post_save は、Model が save メソッドを呼び出す前と後にトリガーされるイベントを表します。これは、によって提供されるトリガーと機能的に同等です。データベース。類似点があります。

シグナルの詳細については、公式ドキュメントを参照してください。シグナルがカウンターにどのようなメリットをもたらすかを見てみましょう。

1. 新しいメッセージが到着したら、カウンターに 1 を加えます

この状況は、Django のシグナルを使用して、わずか数行のコードで実装することができます。

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 シグナルに基づいているため、コード内のどこかで生の 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)

この時点で、通知の削除イベントによってカウンターも通常どおり更新されます。

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 の未読メッセージ オブジェクトがあるとします。この時点で、2 つのリクエストを同時に受信すると、この通知を既読としてマークする必要があります。

# 因为两个并发的请求，假设这两个方法几乎同时被调用
NotificationController(user_id).mark_as_readed(100)
NotificationController(user_id).mark_as_readed(100)

明らかに、これら 2 つのメソッドはこの通知を既読としてマークします。同時実行の場合、notification.has_readed のようなチェックが正しく機能しないため、カウンタは誤って -1 2 回になりますが、実際にはリクエストを 1 つ読みます。

それでは、この問題をどうやって解決すればいいのでしょうか?

基本的に、同時リクエストによって引き起こされるデータ競合を解決する方法は 1 つだけあり、比較的単純な 2 つの解決策が紹介されています。

データベース クエリの更新に選択を使用します

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中来定义执行。

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

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