Python的Django框架中消息通知的计数器实现教程

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

class UserNotificationsCount(models.Model):
 """这个Model保存着每一个用户的未读消息数目"""

 user_id = models.IntegerField(primary_key=True)
 unread_count = models.IntegerField(default=0)

 def __str__(self):
 return '' % (self.user_id, self.unread_count)

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)

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)

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)

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

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)

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)

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)

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中那些没有被回写
# 到数据库的缓冲区数据。在这里代码就省略了

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