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 &#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 は、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 の未読メッセージ オブジェクトがあるとします。この時点で、この通知を既読としてマークする必要があります:

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

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

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

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

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

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

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

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