zoukankan      html  css  js  c++  java
  • 利用django信号实现计数功能

    本文主要知识点:

      1.使用Django的signals来获取Model的新建/删除操作更新
      2.使用数据库的select for update来正确处理并发的数据库操作
      3.使用redis的sorted set来缓存计数器的修改操作

    一,起始

      我们要实现消息未读记录,常规做法:

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

      当你的Notification表比较小的时候,这样的方式没有任何问题。但是随着业务的增大,表中的数据可能有上亿条记录。那么随之用户的未读消息数,也会有所提高。

      这时候,你就需要实现一个计数器,让这个计数器来统计每个用户的未读消息数,这样 比起之前的 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)

      我们为每个注册的用户提供一条对应的记录,来保存用户的未读消息数。

      那么重点就是,我们如何知道什么时候应该更新unread_count呢?

      这就要说到Django给我提供的强大的信号机制,从而方便我们实时检测计数器。

      为了实时更新我们的计数器,我们必须做到以下几点:

        1.当有新的未读消息过来的时候,计数器+1

        2.当消息被异常删除时,如果关联的消息未读,为计数器-1

        3.当阅读完一个新消息的时候,计数器-1

       django.db.models.signals.pre_save & django.db.models.signals.post_save 表示的是 某个Model调用save方法之前和之后会触发的事件

       现在,我们利用Django提供的两个信号机制来实现我们的实时更新计数器。

      1.当有新的未读消息过来的时候,计数器+1

    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。

      2.当消息被异常删除时,如果关联的消息为未读,为计数器 -1

    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)

      显而易见的,这两次方法都会成功的标记这条通知为已读,因为在并发的情况下, if notification.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中来定义执行。

      

  • 相关阅读:
    Spark-sql windows 下 执行错误.
    notepad ++ 注册表
    log4j 配置文件 示例
    linux 查看 进程 内存占用
    spring boot 常见错误解决
    python 轻量 web 框架 Bottle 使用
    Spring cloud eureka 添加 spring-security
    vue can‘ not resolver sass-loader 的 解决办法。
    外国人眼中的珍珠奶茶是啥?
    75.2亿美元:诺基亚、微软终于在一起
  • 原文地址:https://www.cnblogs.com/vipchenwei/p/8581440.html
Copyright © 2011-2022 走看看