如何用Django ORM高效获取每个Dude的最新3条PhoneNumber记录？

你的问题是典型的**Top N per Group（每组取前N条）**场景，原方案的N+1查询在3000万条数据的量级下完全不可行——170个Dude就要触发至少171次数据库查询，还得在Python里循环处理大量数据，CPU和内存消耗都会爆炸。

下面是基于Django ORM的最优实现，只需要1次数据库查询，把大部分计算压力交给数据库（数据库天生擅长处理大数据量的排序和筛选）：

方案一：窗口函数（推荐，Django 2.0+ 支持） #

从Django 2.0开始，ORM支持数据库窗口函数，这是处理Top N per Group最高效的方式，直接在数据库层面完成行号分配和筛选，只返回你需要的510条数据（170*3）。

from django.db.models import Window, F
from django.db.models.functions import RowNumber

# 1. 给每个Dude的PhoneNumber按date倒序分配行号，筛选出行号<=3的记录
# 同时用select_related预加载dude和business，避免后续关联查询
top_phones = PhoneNumber.objects.annotate(
    row_num=Window(
        expression=RowNumber(),
        partition_by=F('dude'),  # 按Dude分组
        order_by=F('date').desc()  # 每个组内按date倒序排序
    )
).filter(row_num__lte=3).select_related('dude', 'business')

# 2. 在Python里快速分组整理成需要的格式
response = []
dude_dict = {}

for phone in top_phones:
    current_dude = phone.dude
    # 如果是第一次处理这个Dude，初始化字典
    if current_dude.id not in dude_dict:
        dude_entry = {
            'name_label': str(current_dude.name),
            'phones': []
        }
        dude_dict[current_dude.id] = dude_entry
        response.append(dude_entry)
    # 添加当前手机号的信息
    phone_entry = {
        'date_added': phone.date.timestamp(),
        'business_label': str(phone.business.name)
    }
    dude_dict[current_dude.id]['phones'].append(phone_entry)

为什么这个方案高效？ #

• 仅1次数据库查询：不管你有多少Dude，所有筛选、排序、关联都在一次查询里完成。
• 最小化数据传输：数据库只返回每个Dude的最新3条PhoneNumber，总共510条数据，而不是3000万条。
• 预加载避免N+1：select_related提前把dude和business的数据一起查出来，不会在循环时触发额外查询。
• Python端轻量处理：用字典分组的时间复杂度是O(M)（M=510），几乎不占CPU。

方案二：兼容旧版数据库（如MySQL 5.7及以下） #

如果你的数据库不支持窗口函数（比如MySQL 5.7），可以用子查询筛选每个Dude的前3个最新日期，再匹配PhoneNumber：

from django.db.models import Subquery, OuterRef

# 1. 子查询获取每个Dude的前3个最新date
top_dates_subquery = PhoneNumber.objects.filter(
    dude=OuterRef('dude')
).order_by('-date').values('date')[:3]

# 2. 筛选出date在子查询结果里的PhoneNumber，同时预加载关联模型
top_phones = PhoneNumber.objects.filter(
    date__in=Subquery(top_dates_subquery)
).select_related('dude', 'business').order_by('dude', '-date')

# 3. 同样用字典分组整理数据，和方案一的第二步完全一致
response = []
dude_dict = {}
for phone in top_phones:
    current_dude = phone.dude
    if current_dude.id not in dude_dict:
        dude_entry = {
            'name_label': str(current_dude.name),
            'phones': []
        }
        dude_dict[current_dude.id] = dude_entry
        response.append(dude_entry)
    phone_entry = {
        'date_added': phone.date.timestamp(),
        'business_label': str(phone.business.name)
    }
    dude_dict[current_dude.id]['phones'].append(phone_entry)

不过这个方案的性能略逊于窗口函数，因为子查询可能会触发数据库的临时表操作，在3000万条数据的场景下，还是优先推荐升级数据库到支持窗口函数的版本（比如MySQL 8.0+）。

对比原方案的性能提升 #

原方案需要至少171次数据库查询，还要循环处理大量数据；新方案只需要1次查询，Python端处理的数据量只有510条，性能提升至少是两个数量级以上，完全能应对3000万条PhoneNumber的场景。

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