Counter for message notifications in Python's Django framework

The beginning of the story: .count()

Suppose you have a Notification Model class that mainly saves all site notifications:

class Notification(models.Model):
  """一个简化过的Notification类，拥有三个字段：

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

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

Of course, at first you will get the number of unread messages of a certain user through such a query:

# 获取ID为3074的用户的未读消息数
Notification.objects.filter(user_id=3074, has_readed=False).count()

When your Notification table is relatively small, there is no problem with this method, but slowly, as the business volume expands. There are hundreds of millions of pieces of data in the message table. Many lazy users have thousands of unread messages.

At this time, you need to implement a counter to count the number of unread messages for each user. In this way, compared to the previous count(), we only need to execute a simple primary key query (or Even better) you can get the real-time number of unread messages.

Better solution: Create a counter
First, let us create a new table to store the number of unread messages for each user.

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)

We provide each registered user with a corresponding UserNotificationsCount record to save his number of unread messages. Every time you get his unread message count, you only need UserNotificationsCount.objects.get(pk=user_id).unread_count.

Next, here comes the focus of the question, how do we know when we should update our counters? Does Django provide any shortcuts in this regard?

Challenge: Update your counter in real time

In order for our counter to work properly, we must update it in real time, which includes:

• When a new unread message comes, the counter is +1

• When the message is deleted abnormally, if the associated message is unread, the counter is -1

• When a new message is read, a counter -1

Let’s address these situations one by one.

Before throwing out the solution, we need to introduce a function in Django: Signals. Signals is an event notification mechanism provided by Django, which allows you to listen to certain custom or preset events. , when these events occur, the implementation-defined methods are called.

For example, django.db.models.signals.pre_save & django.db.models.signals.post_save represent events that will be triggered before and after a Model calls the save method. It is the same as the trigger provided by Database. There is some similarity in functionality.

For more information about Signals, please refer to the official documentation. Let’s take a look at what benefits Signals can bring to our counters.

1. When a new message comes in, add 1 to the counter

This situation should be the best to handle. Using Django's Signals, just short In just a few lines of code, we can update the counter in this case:

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)

In this way, whenever you use Notification.create or .save() When a new notification is created using a method like this, our NotificationController will be notified and the counter will be +1.

But please note that because our counters are based on Django signals, if you use raw sql somewhere in your code and do not add new notifications through the Django ORM method, our counters will not get Notifications, so it is best to standardize all new notification creation methods, such as using the same API.

2. When a message is deleted abnormally, if the associated message is unread, the counter is -1

With the first experience, this situation It is relatively simple to handle. You only need to monitor the post_delete signal of Notification. The following is an example code:

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)

At this point, the deletion of Notification Events can also update our counters normally.

3. When reading a new message, the counter is -1

Next, when the user reads an unread message, we also need to update Our unread message counter. You might say, what's so difficult about this? Can I just update my counter manually in my method of reading messages?

For example:

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)

Through some simple tests, you may feel that your counter works very well, but, like this There is a very fatal problem with the implementation method. This method cannot handle concurrent requests normally.

For example, you have an unread message object with an ID of 100. At this time, two requests come in at the same time. You must mark this notification as read:

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

Obviously, both methods will successfully mark this notification as read, because in the case of concurrency, checks like if notification.has_readed cannot work properly, so our counter will be wrongly -1 twice, but in fact we only read one request.

So, how to solve this problem?

Basically, there is only one way to solve data conflicts caused by concurrent requests: locking. Two relatively simple solutions are introduced:

使用 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中文网！