이야기의 시작: .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는 이와 관련하여 지름길을 제공합니까?
과제: 실시간 카운터 업데이트
카운터가 제대로 작동하려면 다음을 포함하여 실시간으로 업데이트해야 합니다.
이러한 상황을 하나씩 해결해 보겠습니다.
솔루션을 버리기 전에 Django의 기능을 소개해야 합니다. Signals는 Django에서 제공하는 이벤트 알림 메커니즘으로, 이러한 이벤트가 발생할 때 특정 사용자 정의 또는 사전 설정 이벤트를 들을 수 있습니다. , 구현에서 정의된 메서드가 호출됩니다.
예를 들어 django.db.models.signals.pre_save & django.db.models.signals.post_save는 Model이 save 메소드를 호출하기 전후에 트리거되는 이벤트를 나타냅니다. 데이터베이스에서 제공하는 트리거 기능에는 일부 유사점이 있습니다.
Signal에 대한 자세한 내용은 공식 문서를 참조하세요. Signals가 카운터에 어떤 이점을 가져올 수 있는지 살펴보겠습니다.
1. 새 메시지가 오면 카운터에 1을 추가합니다
이 상황은 Django의 Signal을 사용하여 처리하는 것이 가장 좋습니다. 몇 줄의 코드만 있으면 이 경우 카운터 업데이트를 구현할 수 있습니다:
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)로그인 후 복사
이러한 방식으로 알림.create 또는 .save()를 사용할 때마다 이와 같은 메서드를 사용하여 새 알림이 생성되면 우리의 NotificationController가 알림을 받고 카운터는 +1이 됩니다.
그러나 우리 카운터는 Django 신호를 기반으로 하기 때문에 코드 어딘가에서 원시 SQL을 사용하고 Django ORM 메서드를 통해 새 알림을 추가하지 않으면 우리 카운터는 알림을 받지 못합니다. 동일한 API를 사용하는 등 모든 새로운 알림 생성 방법을 표준화하는 것이 가장 좋습니다.
2. 메시지가 비정상적으로 삭제된 경우 관련 메시지를 읽지 않은 경우 카운터는 -1입니다
첫 번째 경험으로도 이 상황은 상대적입니다. 다음은 알림의 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)
로그인 후 복사
분명히 두 가지 방법 모두 성공적으로 이 알림을 읽음으로 표시합니다. 왜냐하면 동시성의 경우 통지.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('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中文网！
![[웹 프런트엔드] Node.js 빠른 시작](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
