Django累计用户活动数据后批量入库的优化方案咨询

• 若站点日活低于1万，直接每30秒更新数据库完全可行，属于过度优化。单条按主键更新的数据库操作耗时通常在1ms以内，即便有1000个同时在线的活跃用户，每秒也仅需处理33次更新，常规MySQL/PostgreSQL实例都能轻松扛住，不需要额外引入复杂度。
• 若活跃用户量超过1万，或者数据库本身负载已经较高，那缓存后批量入库是合理的优化方向。

你遇到的多进程无法共享内存、进程退出需要刷数据的问题，本质是不要把计数存在Web worker的进程内存里，换用独立的共享缓存层即可，Redis是这个场景最适配的选型，实现逻辑非常简单：

核心实现步骤 #

1. 安装django-redis并配置Django缓存后端为Redis，配置后可以直接调用Django自带的缓存接口操作Redis，不需要额外写Redis连接逻辑。
2. 收到心跳请求时，直接对Redis中对应用户+日期的key执行增量操作，代码示例：

from django.core.cache import cache
from django.utils import timezone
from django.http import JsonResponse

def activity_heartbeat(request):
    # 仅处理登录用户的请求，未登录可按业务逻辑过滤
    if not request.user.is_authenticated:
        return JsonResponse({"status": "fail"}, status=401)
    user_id = request.user.id
    today_str = timezone.now().strftime("%Y%m%d")
    cache_key = f"user_activity:{user_id}:{today_str}"
    # 原子累加30秒
    cache.incr(cache_key, 30)
    # 首次写入时设置过期时间，避免无效key长期占用内存
    if cache.ttl(cache_key) < 0:
        cache.expire(cache_key, 60 * 60 * 48)
    return JsonResponse({"status": "ok"})

3. 每10分钟执行一次批量入库逻辑，你可以选自己熟悉的定时方案：
   • 轻量方案：写一个Django自定义管理命令，用Linux crontab每10分钟触发一次
   • 如果你已经在用Celery做异步任务，用Celery Beat定时触发即可
4. 批量入库的代码示例（自定义管理命令）：

from django.core.management.base import BaseCommand
from django.core.cache import cache
from django.utils import timezone
from yourapp.models import UserActivity
import re

class Command(BaseCommand):
    def handle(self, *args, **options):
        today_str = timezone.now().strftime("%Y%m%d")
        # 匹配当天所有用户的活动计数key
        activity_keys = cache.keys(f"user_activity:*:{today_str}")
        update_list = []
        for key in activity_keys:
            key_match = re.match(r"user_activity:(\d+):\d+", key)
            if not key_match:
                continue
            user_id = int(key_match.group(1))
            total_seconds = int(cache.get(key, 0))
            if total_seconds <= 0:
                continue
            # 构造待更新/插入的对象
            update_list.append(UserActivity(
                user_id=user_id,
                date=timezone.now().date(),
                duration=total_seconds
            ))
        # 批量写入，冲突时更新时长即可
        if update_list:
            UserActivity.objects.bulk_create(
                update_list,
                update_conflicts=True,
                unique_fields=["user_id", "date"],
                update_fields=["duration"]
            )
        # 写入完成后删除已同步的key，也可以等过期自动删除
        cache.delete_many(activity_keys)

crontab配置参考：

*/10 * * * * /你的虚拟环境路径/bin/python /你的项目路径/manage.py sync_activity_to_db >> /日志路径/sync_activity.log 2>&1

方案优势 #

• 天然解决多worker共享计数的问题：Redis是独立的服务，所有Web worker的请求都操作同一个Redis实例，计数不会错乱
• 不需要处理worker进程退出的逻辑：计数都存在Redis中，Web worker重启、崩溃都不会丢失未同步的计数，完全符合你接受最多丢失10分钟数据的要求
• 性能足够：单Redis实例每秒可以扛至少10万次增量操作，远高于Web服务的处理上限

内容的提问来源于stack exchange，提问作者Ilya