perf(topics): 新建/删除/重命名话题同样走 spawn_blocking 统一提速

- 抽取 list_topic_summaries_blocking 公共 helper(list_topics + build_topic_summaries 在 blocking 线程池执行)
- create/delete/rename 三个话题命令返回侧边栏刷新列表时统一走 helper,DB 查询不再阻塞 tokio worker
- 结合增量统计列,四个话题命令的列表构建均从 O(全部消息) 降为 O(话题数)
This commit is contained in:
oudecheng 2026-08-18 23:06:19 +08:00
parent f711948902
commit 985904b95e
4 changed files with 48 additions and 40 deletions

View File

@ -1,7 +1,6 @@
use crate::command::Command; use crate::command::Command;
use crate::command::context::CommandContext; use crate::command::context::CommandContext;
use crate::command::handler::{CommandHandler, CommandMetadata}; use crate::command::handler::{CommandHandler, CommandMetadata};
use crate::command::handlers::list_topics::build_topic_summaries;
use crate::command::response::{CommandError, CommandResponse, MessageKind}; use crate::command::response::{CommandError, CommandResponse, MessageKind};
use crate::gateway::session::SessionManager; use crate::gateway::session::SessionManager;
use crate::storage::SessionStore; use crate::storage::SessionStore;
@ -82,12 +81,12 @@ async fn handle_delete_topic(
.map_err(|e| CommandError::new("DELETE_TOPIC_ERROR", e.to_string()))?; .map_err(|e| CommandError::new("DELETE_TOPIC_ERROR", e.to_string()))?;
// 查询更新后的话题列表,返回给前端刷新侧边栏 // 查询更新后的话题列表,返回给前端刷新侧边栏
let topics = handler let topic_summaries =
.store crate::command::handlers::list_topics::list_topic_summaries_blocking(
.list_topics(session_id) handler.store.clone(),
.map_err(|e| CommandError::new("LIST_TOPICS_ERROR", e.to_string()))?; session_id,
)
let topic_summaries = build_topic_summaries(handler.store.as_ref(), topics)?; .await?;
let topics_json = serde_json::to_string(&topic_summaries) let topics_json = serde_json::to_string(&topic_summaries)
.map_err(|e| CommandError::new("SERIALIZE_ERROR", e.to_string()))?; .map_err(|e| CommandError::new("SERIALIZE_ERROR", e.to_string()))?;

View File

@ -97,6 +97,28 @@ pub fn build_topic_summaries(
Ok(summaries) Ok(summaries)
} }
/// 在 blocking 线程池中执行 list_topics + build_topic_summaries。
///
/// 同步 rusqlite 查询不得直接跑在 tokio worker 上,否则大库查询会饿死
/// 同运行时上的其他任务。list / create / delete / rename 四个话题命令
/// 都返回完整的 TopicSummary 列表供前端刷新侧边栏,统一走本 helper。
pub async fn list_topic_summaries_blocking(
store: Arc<SessionStore>,
session_id: &str,
) -> Result<Vec<TopicSummary>, CommandError> {
let session_id_bg = session_id.to_string();
tokio::task::spawn_blocking(
move || -> Result<Vec<TopicSummary>, CommandError> {
let topics = store
.list_topics(&session_id_bg)
.map_err(|e| CommandError::new("LIST_TOPICS_ERROR", e.to_string()))?;
build_topic_summaries(store.as_ref(), topics)
},
)
.await
.map_err(|e| CommandError::new("LIST_TOPICS_ERROR", e.to_string()))?
}
/// 列出 Session 的 Topics 命令处理器 /// 列出 Session 的 Topics 命令处理器
pub struct ListTopicsCommandHandler { pub struct ListTopicsCommandHandler {
store: Arc<SessionStore>, store: Arc<SessionStore>,
@ -139,20 +161,8 @@ async fn handle_list_topics(
session_id: String, session_id: String,
ctx: CommandContext, ctx: CommandContext,
) -> Result<CommandResponse, CommandError> { ) -> Result<CommandResponse, CommandError> {
// 同步 rusqlite 查询移入 blocking 线程池,避免大库查询阻塞 tokio worker let summaries =
// 饿死同连接上的其他 WS 命令(话题列表刷新是高频操作)。 list_topic_summaries_blocking(handler.store.clone(), &session_id).await?;
let store = handler.store.clone();
let session_id_bg = session_id.clone();
let summaries = tokio::task::spawn_blocking(
move || -> Result<Vec<TopicSummary>, CommandError> {
let topics = store
.list_topics(&session_id_bg)
.map_err(|e| CommandError::new("LIST_TOPICS_ERROR", e.to_string()))?;
build_topic_summaries(store.as_ref(), topics)
},
)
.await
.map_err(|e| CommandError::new("LIST_TOPICS_ERROR", e.to_string()))??;
let topics_json = serde_json::to_string(&summaries) let topics_json = serde_json::to_string(&summaries)
.map_err(|e| CommandError::new("SERIALIZE_ERROR", e.to_string()))?; .map_err(|e| CommandError::new("SERIALIZE_ERROR", e.to_string()))?;

View File

@ -1,7 +1,6 @@
use crate::command::Command; use crate::command::Command;
use crate::command::context::CommandContext; use crate::command::context::CommandContext;
use crate::command::handler::{CommandHandler, CommandMetadata}; use crate::command::handler::{CommandHandler, CommandMetadata};
use crate::command::handlers::list_topics::build_topic_summaries;
use crate::command::response::{CommandError, CommandResponse, MessageKind}; use crate::command::response::{CommandError, CommandResponse, MessageKind};
use crate::storage::SessionStore; use crate::storage::SessionStore;
use async_trait::async_trait; use async_trait::async_trait;
@ -84,11 +83,12 @@ async fn handle_rename_topic(
// 标题未变化时直接返回当前列表,避免无意义写入 // 标题未变化时直接返回当前列表,避免无意义写入
if old_display == trimmed_title { if old_display == trimmed_title {
let topics = handler let topic_summaries =
.store crate::command::handlers::list_topics::list_topic_summaries_blocking(
.list_topics(session_id) handler.store.clone(),
.map_err(|e| CommandError::new("LIST_TOPICS_ERROR", e.to_string()))?; session_id,
let topic_summaries = build_topic_summaries(handler.store.as_ref(), topics)?; )
.await?;
let topic_summaries_json = serde_json::to_string(&topic_summaries) let topic_summaries_json = serde_json::to_string(&topic_summaries)
.map_err(|e| CommandError::new("SERIALIZE_ERROR", e.to_string()))?; .map_err(|e| CommandError::new("SERIALIZE_ERROR", e.to_string()))?;
@ -110,12 +110,12 @@ async fn handle_rename_topic(
.map_err(|e| CommandError::new("RENAME_TOPIC_ERROR", e.to_string()))?; .map_err(|e| CommandError::new("RENAME_TOPIC_ERROR", e.to_string()))?;
// 查询更新后的话题列表,返回给前端刷新侧边栏 // 查询更新后的话题列表,返回给前端刷新侧边栏
let topics = handler let topic_summaries =
.store crate::command::handlers::list_topics::list_topic_summaries_blocking(
.list_topics(session_id) handler.store.clone(),
.map_err(|e| CommandError::new("LIST_TOPICS_ERROR", e.to_string()))?; session_id,
)
let topic_summaries = build_topic_summaries(handler.store.as_ref(), topics)?; .await?;
let topic_summaries_json = serde_json::to_string(&topic_summaries) let topic_summaries_json = serde_json::to_string(&topic_summaries)
.map_err(|e| CommandError::new("SERIALIZE_ERROR", e.to_string()))?; .map_err(|e| CommandError::new("SERIALIZE_ERROR", e.to_string()))?;

View File

@ -1,7 +1,6 @@
use crate::command::Command; use crate::command::Command;
use crate::command::context::CommandContext; use crate::command::context::CommandContext;
use crate::command::handler::{CommandHandler, CommandMetadata}; use crate::command::handler::{CommandHandler, CommandMetadata};
use crate::command::handlers::list_topics::build_topic_summaries;
use crate::command::response::{CommandError, CommandResponse, MessageKind}; use crate::command::response::{CommandError, CommandResponse, MessageKind};
use crate::gateway::session::SessionManager; use crate::gateway::session::SessionManager;
use crate::storage::SessionStore; use crate::storage::SessionStore;
@ -104,12 +103,12 @@ async fn handle_create_session(
} }
// Query the full topic list so the frontend sidebar can update // Query the full topic list so the frontend sidebar can update
let topics = handler let topic_summaries =
.store crate::command::handlers::list_topics::list_topic_summaries_blocking(
.list_topics(session_id) handler.store.clone(),
.map_err(|e| CommandError::new("LIST_TOPICS_ERROR", e.to_string()))?; session_id,
)
let topic_summaries = build_topic_summaries(handler.store.as_ref(), topics)?; .await?;
let topics_json = serde_json::to_string(&topic_summaries) let topics_json = serde_json::to_string(&topic_summaries)
.map_err(|e| CommandError::new("SERIALIZE_ERROR", e.to_string()))?; .map_err(|e| CommandError::new("SERIALIZE_ERROR", e.to_string()))?;