如何使用缓存加速Django中层级关联模型的堆叠查询速度

方案1：预计算结果持久化到数据库（推荐，符合你的需求） #

这个方案直接在Entry写入/关联Data变动时自动计算generateFields结果并存入字段，查询时直接读取，完全避免运行时计算和反向查询开销。

1. 给Entry模型新增JSON缓存字段

from django.db import models
import math

class Entry(models.Model):
    time = models.DateTimeField()
    sensor = models.ForeignKey(Sensor, on_delete=models.CASCADE, related_name="entriesList")
    # 新增缓存字段，存储generateFields的计算结果
    cached_fields = models.JSONField(null=True, blank=True)
    
    def save(self, *args, **kwargs):
        # 新增时如果没有缓存就预计算（更新Data时会单独触发更新）
        if not self.cached_fields:
            self.cached_fields = self._generate_fields()
        super().save(*args, **kwargs)
    
    # 把原来的generateFields拆成内部计算方法和对外读取方法
    def _generate_fields(self, field=None):
        # 原来的generateFields逻辑保持不变
        if field is None:
            field_set = self.dataFields.all()
        else:
            field_set = self.dataFields.filter(key=field)

        output = {"timestamp": self.time.timestamp()}

        for key, value in field_set.values_list("key", "value"):
            try:
                floated = float(value)
                isnan = math.isnan(floated)
            except (TypeError, ValueError):
                continue

            if isnan:
                return None
            else:
                output[key] = floated

        return output
    
    def generateFields(self, field=None):
        # 优先读缓存，没有缓存再实时计算
        if field is None and self.cached_fields is not None:
            return self.cached_fields
        return self._generate_fields(field)

2. 处理Data变动时的缓存更新
   因为Data的增删改会影响对应Entry的计算结果，所以需要在Data模型的保存/删除逻辑里触发对应Entry的缓存更新：

class Data(models.Model):
    key = models.CharField(max_length=50)
    value = models.CharField(max_length=50)
    entry = models.ForeignKey(Entry, on_delete=models.CASCADE, related_name="dataFields")

    def save(self, *args, **kwargs):
        super().save(*args, **kwargs)
        # Data保存后更新对应Entry的缓存
        self.entry.cached_fields = self.entry._generate_fields()
        self.entry.save(update_fields=['cached_fields'])
    
    def delete(self, *args, **kwargs):
        entry = self.entry
        super().delete(*args, **kwargs)
        # Data删除后更新对应Entry的缓存
        entry.cached_fields = entry._generate_fields()
        entry.save(update_fields=['cached_fields'])

如果有批量操作Data的场景，需要额外处理批量操作后的缓存更新，也可以定期跑脚本校准缓存保证一致性。

方案2：无需改表的查询侧优化（见效最快） #

你当前性能问题的核心是N+1查询：循环5000个Entry时，每个Entry都会单独触发一次dataFields的数据库查询，累计产生5000+次查询请求，开销极大。用prefetch_related预取所有关联的Data，只需要2次查询就能拿到所有数据，性能可以直接提升10倍以上。
修改Sensor的generateData方法即可：

from django.db.models import Prefetch
from datetime import datetime

def generateData(self, fromTime, toTime, field=None):
    fromTime = datetime.fromtimestamp(fromTime)
    toTime = datetime.fromtimestamp(toTime)
    # 预取关联的dataFields，避免循环时重复查询
    if field is None:
        entries = self.entriesList.filter(time__range=(toTime, fromTime)).order_by("time").prefetch_related("dataFields")
    else:
        # 指定查询字段时，只预取符合条件的Data，减少数据传输量
        entries = self.entriesList.filter(time__range=(toTime, fromTime)).order_by("time").prefetch_related(
            Prefetch("dataFields", queryset=Data.objects.filter(key=field))
        )
    output = []
    for entry in entries:
        # 这里的dataFields已经预取到内存，不会再查数据库
        value = entry.generateFields(field)
        if value is not None:
            output.append(value)
    return output

方案选择建议 #

• 如果你的业务是读多写少，优先选方案1，写入时一次计算，所有查询直接读缓存，性能提升最大，符合你测算的25-50倍的提升预期。
• 如果业务写入非常频繁，或者不方便做表结构迁移，优先选方案2，无需修改代码逻辑之外的内容，改完即可生效。

内容的提问来源于stack exchange，提问作者Zaid Al Shattle