列表页别逐条查：我在 Rust CRM 里用 is_in + HashMap 干掉 N+1

写后台系统的时候，有一种性能问题特别隐蔽：列表页只是多显示几个名称，数据库查询次数却突然翻倍。

比如订单列表。

订单表里有customer_uuid，页面上要展示客户名；订单还可能来自服务请求，页面上又要展示服务项目名。需求看起来很普通，于是很容易写出这种代码：

// 伪代码：别这么写letorders=OrderEntity::find().offset(...).limit(...).all(db).await?;fororderinorders{letcustomer=ContactEntity::find().filter(ContactColumn::ContactUuid.eq(order.customer_uuid)).one(db).await?;letrequest=ServiceRequestEntity::find().filter(ServiceRequestColumn::Uuid.eq(order.request_id)).one(db).await?;// 再根据 request.service_catalog_uuid 查服务项目...}

这段代码最大的问题不是“慢”，而是慢得很稳定地不可控。

分页 20 条时，看起来还能忍；分页 100 条、字段再多几个、网络抖一下，列表页就开始卡。更麻烦的是，这类问题经常在开发环境看不出来，因为本地数据太少、数据库就在旁边。

所以我后来在 Pico-CRM 的查询层定了一个很朴素的习惯：

列表页里不要在for/map里查数据库。

一、N+1 在订单列表里长什么样

Pico-CRM 是一个家政服务 CRM，核心流程是：

客户 -> 服务需求 -> 订单 -> 排班 -> 完工

订单列表的主数据来自orders表，但页面并不只看订单本身，还要展示：

• 客户姓名：来自contacts
• 服务项目：订单关联服务请求，再由服务请求关联service_catalogs
• 订单状态、时间、金额等订单字段

也就是说，一行订单 DTO 里混了几类展示字段。

如果按照逐行查询来算，分页 20 条时，最差情况大概是：

1 次订单分页查询 + 20 次客户查询 + 20 次服务请求查询 + 20 次服务项目查询 = 61 次查询

分页大小一改，查询次数跟着涨。

这就是典型的 N+1。

这里的 “N” 不是一个抽象概念，而是后台列表页的分页条数。产品觉得 20 条太少，想改成 50 条，接口压力就会马上变大。

二、先分页，再批量补字段

Pico-CRM 里订单列表的实现放在：

backend/src/infrastructure/queries/crm/order_query_impl.rs

真实流程是这样：

lettotal=select.clone().count(txn).await.map_err(|e|format!("query orders count error: {}",e))?;letmodels=select.order_by_desc(Column::InsertedAt).offset(Some((query.page-1)*query.page_size)).limit(Some(query.page_size)).all(txn).await.map_err(|e|format!("query orders error: {}",e))?;

第一步还是正常的主表分页：先查总数，再查当前页数据。

重点在第二步：只对当前页数据收集关联 ID。

letcustomer_ids:HashSet<Uuid>=models.iter().filter_map(|model|model.customer_uuid).collect();letrequest_ids:HashSet<Uuid>=models.iter().filter_map(|model|model.request_id).collect();

这里用HashSet有两个好处：

• 自动去重，避免同一个客户被查多次
• 当前页没有关联数据时，可以直接跳过后续查询

然后客户信息一次查完：

letmutcustomer_map:HashMap<Uuid,String>=HashMap::new();if!customer_ids.is_empty(){letcustomers=ContactEntity::find().filter(ContactColumn::ContactUuid.is_in(customer_ids.clone())).all(txn).await.map_err(|e|format!("query customers error: {}",e))?;forcustomerincustomers{customer_map.insert(customer.contact_uuid,customer.user_name);}}

这段代码的核心就两件事：

WHERE contact_uuid IN (...) 查询结果塞进 HashMap<Uuid, String>

后面回填 DTO 时，直接按 ID 查内存 map：

ifletSome(customer_uuid)=customer_uuid{view.customer_name=customer_map.get(&customer_uuid).cloned();}

查询次数不会随着订单条数线性增长。

三、关联链路多一层，也不要回到逐条查

客户名只隔一张表，比较简单。

服务项目名稍微绕一点：订单里不是直接存service_catalog_uuid，而是存了request_id。要拿服务项目名，需要走：

orders.request_id -> service_requests.uuid -> service_requests.service_catalog_uuid -> service_catalogs.name

这时很多人会下意识写成循环里查两次。

Pico-CRM 里还是同一个套路，只是拆成两段批量查询。

先批量查服务请求，并建立request_id -> service_catalog_uuid的映射：

letmutrequest_service_catalog_map:HashMap<Uuid,Option<Uuid>>=HashMap::new();if!request_ids.is_empty(){letrequests=ServiceRequestEntity::find().filter(ServiceRequestColumn::Uuid.is_in(request_ids.clone())).all(txn).await.map_err(|e|format!("query service requests error: {}",e))?;forrequestinrequests{request_service_catalog_map.insert(request.uuid,request.service_catalog_uuid,);}}

再从这些请求里收集服务项目 ID：

letservice_catalog_ids:HashSet<Uuid>=request_service_catalog_map.values().filter_map(|value|*value).collect();

最后批量查服务目录：

letmutservice_catalog_map:HashMap<Uuid,String>=HashMap::new();if!service_catalog_ids.is_empty(){letcatalogs=ServiceCatalogEntity::find().filter(ServiceCatalogColumn::Uuid.is_in(service_catalog_ids.clone())).all(txn).await.map_err(|e|format!("query service catalogs error: {}",e))?;forcatalogincatalogs{service_catalog_map.insert(catalog.uuid,catalog.name);}}

回填时再把两层 map 串起来：

ifletSome(request_id)=request_id{ifletSome(service_catalog_uuid)=request_service_catalog_map.get(&request_id).and_then(|value|*value){view.service_catalog_uuid=Some(service_catalog_uuid.to_string());view.service_catalog_name=service_catalog_map.get(&service_catalog_uuid).cloned();}}

这样一来，订单列表的查询次数变成了稳定的几次：

1 次订单 count 1 次订单分页 1 次客户批量查询 1 次服务请求批量查询 1 次服务目录批量查询

分页 20 条是这些查询，分页 100 条还是这些查询。

当然，IN (...)的参数数量也不是无限的。但后台分页列表通常是几十到一两百条，放在这个场景里，比每行查一次稳得多。

四、为什么不直接写一个大 join

这个问题很自然：既然都是查关联字段，为什么不直接 SQL join？

我的答案是：能 join，但不是所有列表页都值得 join 成一条 SQL。

订单列表这里有几个特点：

• 主表必须先按订单条件分页
• 客户名、服务项目名只是展示字段
• 服务项目隔了一层服务请求
• DTO 回填时还要保留一些Option语义
• 查询层本来就是读模型适配层，不是领域规则所在

如果用 join，也能做。但 SQL 会越来越长，SeaORM 查询表达式也会更重。后面再加一个展示字段，比如创建人、派工人、服务人员，join 链会继续膨胀。

而现在这种写法的结构很清楚：

主表分页 -> 收集本页关联 ID -> 按表批量查询 -> HashMap 回填 DTO

它牺牲了一点“单 SQL 的纯粹”，换来的是代码可读性和稳定查询次数。

我不是反对 join。像强过滤、强排序、必须由数据库完成聚合的场景，就应该让 SQL 来做。

但对于后台列表页的展示字段补齐，尤其是“分页以后补名称”的场景，内存 join 很好用。

五、服务请求列表也用了同一套模式

同样的实现也出现在：

backend/src/infrastructure/queries/crm/service_request_query_impl.rs

服务请求列表要展示三类名称：

• 客户姓名：contacts
• 创建人姓名：users
• 服务项目名：service_catalogs

代码先从当前页模型里收集三组 ID：

letcustomer_ids:HashSet<Uuid>=models.iter().map(|model|model.customer_uuid).collect();letuser_ids:HashSet<Uuid>=models.iter().map(|model|model.creator_uuid).collect();letservice_catalog_ids:HashSet<Uuid>=models.iter().filter_map(|model|model.service_catalog_uuid).collect();

然后分别用is_in批量查三张表，构建三个 map：

letmutcontact_map:HashMap<Uuid,String>=HashMap::new();letmutuser_map:HashMap<Uuid,String>=HashMap::new();letmutservice_catalog_map:HashMap<Uuid,String>=HashMap::new();

最后统一回填：

view.contact_name=contact_map.get(&customer_uuid).cloned();view.creator_name=user_map.get(&creator_uuid).cloned();view.service_catalog_name=service_catalog_uuid.and_then(|uuid|service_catalog_map.get(&uuid).cloned());

这就是我比较喜欢的代码形态：读起来很笨，但每一步都明确。

谁负责分页，谁负责批量查，谁负责 DTO 字段回填，都能一眼看出来。

六、员工绩效统计：不只是补名称，聚合也能这么做

is_in + HashMap不只适合补名称。

Pico-CRM 的员工列表里有一段绩效统计，代码在：

backend/src/infrastructure/queries/identity/user_query_impl.rs

它要算每个家政人员的：

• 已完成服务次数
• 评价数
• 平均评分
• 售后数
• 投诉数
• 退款数
• 返工数

这里如果逐个员工查一遍排班、评价、售后，查询次数会更夸张。

所以实现里先对当前页员工 ID 批量查已完成排班：

letschedule_items=ScheduleEntity::find().filter(ScheduleColumn::AssignedUserUuid.is_in(user_ids.iter().copied())).filter(ScheduleColumn::Status.eq("done")).all(txn).await.map_err(|e|format!("query user completed schedules error: {}",e))?;

评价也批量查：

letfeedback_items=OrderFeedbackEntity::find().filter(OrderFeedbackColumn::UserUuid.is_in(user_ids.iter().copied())).all(txn).await.map_err(|e|format!("query user feedback stats error: {}",e))?;

然后把订单和员工关系做成 map：

letschedule_order_map=schedule_items.iter().map(|item|(item.order_uuid,item.assigned_user_uuid)).collect::<HashMap<_,_>>();

售后表是按订单关联的，于是先从schedule_order_map收集订单 ID，再批量查售后：

letorder_ids=schedule_order_map.keys().copied().collect::<Vec<_>>();letafter_sales_cases=iforder_ids.is_empty(){Vec::new()}else{AfterSalesEntity::find().filter(AfterSalesColumn::OrderUuid.is_in(order_ids.clone())).all(txn).await.map_err(|e|format!("query user after sales cases error: {}",e))?};

统计结果则落到：

letmutstats_map=HashMap::<Uuid,UserPerformanceStats>::new();

后面不再查数据库，而是在内存里累加：

foriteminschedule_items{letentry=stats_map.entry(item.assigned_user_uuid).or_default();entry.completed_service_count+=1;}

平均评分也是先收集，再统一计算：

letmutrating_sums=HashMap::<Uuid,(i64,u64)>::new();foriteminfeedback_items{letSome(user_uuid)=item.user_uuidelse{continue;};letentry=stats_map.entry(user_uuid).or_default();entry.feedback_count+=1;ifletSome(rating)=item.rating{letrating_entry=rating_sums.entry(user_uuid).or_insert((0,0));rating_entry.0+=ratingasi64;rating_entry.1+=1;}}

这段和订单列表本质一样：

当前页主实体 -> 收集关联键 -> 批量查相关表 -> 用 HashMap 归并结果

只不过订单列表是回填展示字段，员工绩效是回填统计字段。

七、这套写法的边界

这个模式很好用，但不是银弹。

我现在大概按这几个标准判断。

适合用is_in + HashMap的场景：

• 主表已经分页
• 关联字段只是展示或轻量统计
• 关联数据量跟分页大小同级
• 业务上允许在应用层组装 DTO
• 查询可读性比“单条 SQL”更重要

更适合 SQL join / group by 的场景：

• 需要按关联表字段过滤
• 需要按关联表字段排序
• 聚合数据量远大于当前页
• 需要数据库利用索引和执行计划完成复杂计算
• 返回数据必须严格由一条一致性查询保证

还有一个细节：Pico-CRM 的这些查询基本都包在with_shared_txn()里，同一个列表查询里的多次读取发生在一个事务上下文中。

这不是为了解决所有一致性问题，而是让一次查询组装过程的数据库访问边界更清晰。

总结

N+1 查询不一定要靠复杂框架解决。

在很多后台系统里，最实用的办法反而是这三步：

HashSet 收 ID SeaORM is_in 批量查 HashMap 回填 DTO

Pico-CRM 里的订单列表、服务请求列表、员工绩效统计都用了这个套路。它没有什么炫技的地方，但能把“分页越大查询越多”的问题，压回到稳定的几次数据库访问。

我现在看列表页接口，第一眼就会搜一件事：

有没有在循环里查数据库。

如果有，基本就该停下来，把 ID 先收集起来。

项目开源在 GitHub，搜Pico-CRM可以看到完整代码。

你们项目里的列表页关联字段，是习惯直接 join，还是应用层批量补字段？评论区聊聊。