perf: 深度优化消息内存占用

- 消息缓冲 Vec → VecDeque：原来溢出时 drain(0..n) 每条消息 O(n)
  搬移，高频消息下产生大量拷贝；环形缓冲均摊 O(1)，并按配置上限
  预分配避免反复扩容
- connection_id/topic/payload 改 Arc<str>：引擎侧主题驻留
  （10k 上限，超出整体重建），同主题数千条记录只存一份；
  消息列表每帧克隆最多 300 条用于渲染，payload 共享后零拷贝
- 单条 payload 历史保留上限 256KB（retained_payload，UTF-8
  边界安全截断），超大报文不再长期占据环形缓冲；消息行显示
  「已截断」徽章；发布记录改用实际发送字节（修 Base64/Hex
  历史内容与真实报文不一致的隐患）
- live 测试适配 Arc<str>；补截断边界单元测试（22 个 lib 测试 +
  2 个 live 测试通过）

gpui-app/src/model.rs

@@ -2,6 +2,8 @@
 //!
 //! 本层不依赖 GPUI，可被持久化层与 MQTT 引擎共用。
 
+use std::sync::Arc;
+
 use serde::{Deserialize, Serialize};
 
 // ─── 连接配置 ────────────────────────────────────────────────────────────────
@@ -410,13 +412,17 @@ impl PayloadFormat {
 
 
 /// 一条收到/发出的消息记录。payload 以 UTF-8 文本保存（非文本按 lossy 显示）。
+///
+/// 内存优化：`connection_id`/`topic` 用 `Arc<str>`——同一连接内主题高度重复，
+/// 驻留后数千条记录只保留一份字符串。
 #[derive(Debug, Clone)]
 pub struct MqttRecord {
     pub seq: u64,
-    pub connection_id: String,
+    pub connection_id: Arc<str>,
     pub direction: Direction,
-    pub topic: String,
+    pub topic: Arc<str>,
-    pub payload: String,
+    /// Arc 共享：消息列表渲染每帧克隆最多数百条记录，零拷贝
+    pub payload: Arc<str>,
     pub qos: u8,
     pub retain: bool,
     /// unix 毫秒
@@ -427,6 +433,29 @@ pub struct MqttRecord {
     pub correlation_data: Option<String>,
     pub message_expiry_interval: Option<u32>,
     pub subscription_identifier: Option<u32>,
+    /// payload 超过 [`MAX_PAYLOAD_RETAIN`] 被截断
+    pub payload_truncated: bool,
+}
+
+/// 消息历史中单条 payload 的保留上限。超限截断，避免个别大报文
+/// 长期占据历史缓冲（MQTTX 亦有类似限制）。
+pub const MAX_PAYLOAD_RETAIN: usize = 256 * 1024;
+
+/// 把原始 payload 字节转成用于历史记录的共享文本，超限截断。
+/// 返回 (Arc<str>, 是否截断)。截断时回退到 UTF-8 字符边界。
+pub fn retained_payload(raw: &[u8]) -> (Arc<str>, bool) {
+    if raw.len() <= MAX_PAYLOAD_RETAIN {
+        (Arc::from(String::from_utf8_lossy(raw).as_ref()), false)
+    } else {
+        let mut end = MAX_PAYLOAD_RETAIN;
+        while end > 0 && (raw[end] & 0xC0) == 0x80 {
+            end -= 1;
+        }
+        (
+            Arc::from(String::from_utf8_lossy(&raw[..end]).as_ref()),
+            true,
+        )
+    }
 }
 
 // ─── 发布参数 / 发布预设 ──────────────────────────────────────────────────────
@@ -863,4 +892,29 @@ mod tests {
         // Subscription::new 同样产出默认启用的订阅
         assert!(Subscription::new("c1", "t", 0).enabled);
     }
+
+    // payload 截断：上限内不截断；超出时截断且不切坏 UTF-8 字符
+    #[test]
+    fn retained_payload_caps_and_preserves_utf8() {
+        let small = b"hello";
+        let (p, trunc) = retained_payload(small);
+        assert_eq!(&*p, "hello");
+        assert!(!trunc);
+
+        // 中文每字符 3 字节，让上限落在字符中间（256K 不是 3 的倍数）
+        let big: Vec<u8> = "中".repeat(MAX_PAYLOAD_RETAIN).into_bytes();
+        let (p, trunc) = retained_payload(&big);
+        assert!(trunc, "超过上限必须截断");
+        assert!(p.len() <= MAX_PAYLOAD_RETAIN);
+        assert_eq!(p.chars().last(), Some('中'), "不能切坏 UTF-8 字符");
+        assert!(
+            big.starts_with(p.as_bytes()),
+            "截断内容必须是原数据前缀"
+        );
+
+        // 恰好上限不截断
+        let exact = vec![b'a'; MAX_PAYLOAD_RETAIN];
+        let (_, trunc) = retained_payload(&exact);
+        assert!(!trunc);
+    }
 }

gpui-app/src/mqtt.rs

@@ -37,11 +37,11 @@ pub enum EngineEvent {
     },
     /// 收到消息
     Message(MqttRecord),
-    /// 一条本地发出的消息（已成功交给客户端）
+    /// 一条本地发出的消息（已成功交给客户端）。payload 为实际发送的字节。
     Published {
         connection_id: String,
         topic: String,
-        payload: String,
+        payload: Vec<u8>,
         qos: u8,
         retain: bool,
         content_type: Option<String>,
@@ -77,6 +77,8 @@ struct ConnHandle {
 pub struct MqttEngine {
     runtime: tokio::runtime::Runtime,
     conns: Mutex<HashMap<String, ConnHandle>>,
+    /// 主题/连接 ID 驻留缓存：同一连接内主题高度重复，驻留让消息记录共享字符串
+    topics: Mutex<HashMap<String, Arc<str>>>,
     tx: EventSender,
     seq: Arc<AtomicU64>,
 }
@@ -96,6 +98,7 @@ impl MqttEngine {
         Arc::new(Self {
             runtime,
             conns: Mutex::new(HashMap::new()),
+            topics: Mutex::new(HashMap::new()),
             tx,
             seq: Arc::new(AtomicU64::new(1)),
         })
@@ -272,6 +275,8 @@ impl MqttEngine {
                 .raw_bytes
                 .clone()
                 .unwrap_or_else(|| params.payload.as_bytes().to_vec());
+            // 实际发送的字节留给 Published 事件（记录需要真实内容）
+            let sent_payload = payload.clone();
             // v5 与 v4 两个 crate 各自定义 ClientError，统一归一化为 String。
             let result: Result<(), String> = match &client {
                 ClientKind::V5(client) => {
@@ -318,7 +323,7 @@ impl MqttEngine {
                     engine.emit(EngineEvent::Published {
                         connection_id,
                         topic: params.topic,
-                        payload: params.payload,
+                        payload: sent_payload,
                         qos: params.qos,
                         retain: params.retain,
                         content_type: params.content_type,
@@ -490,6 +495,22 @@ impl MqttEngine {
         rx
     }
 
+    /// 驻留字符串（主题/连接 ID）：同一内容全进程只保留一份 `Arc<str>`。
+    /// 缓存有上限，异常流量下整体清空重建，避免无限增长。
+    pub(crate) fn intern(&self, s: &str) -> Arc<str> {
+        const INTERN_CAP: usize = 10_000;
+        let mut cache = self.topics.lock().unwrap();
+        if let Some(arc) = cache.get(s) {
+            return arc.clone();
+        }
+        if cache.len() >= INTERN_CAP {
+            cache.clear();
+        }
+        let arc: Arc<str> = Arc::from(s);
+        cache.insert(s.to_string(), arc.clone());
+        arc
+    }
+
     pub fn test_connection(
         self: &Arc<Self>,
         cfg: ConnectionConfig,
@@ -942,14 +963,15 @@ async fn test_handshake(cfg_in: &ConnectionConfig) -> Result<(), String> {
 
 // ─── 事件循环 ────────────────────────────────────────────────────────────────
 
-fn v5_record(seq: u64, conn_id: &str, p: &V5Publish) -> MqttRecord {
+fn v5_record(engine: &MqttEngine, seq: u64, conn_id: &str, p: &V5Publish) -> MqttRecord {
     let props = p.properties.as_ref();
+    let (payload, payload_truncated) = crate::model::retained_payload(&p.payload);
     MqttRecord {
         seq,
-        connection_id: conn_id.to_string(),
+        connection_id: engine.intern(conn_id),
         direction: Direction::Received,
-        topic: String::from_utf8_lossy(&p.topic).to_string(),
+        topic: engine.intern(&String::from_utf8_lossy(&p.topic)),
-        payload: String::from_utf8_lossy(&p.payload).to_string(),
+        payload,
         qos: v5_qos_u8(p.qos),
         retain: p.retain,
         timestamp: now_ms(),
@@ -966,6 +988,7 @@ fn v5_record(seq: u64, conn_id: &str, p: &V5Publish) -> MqttRecord {
         message_expiry_interval: props.and_then(|p| p.message_expiry_interval),
         subscription_identifier: props
             .and_then(|p| p.subscription_identifiers.first().map(|v| *v as u32)),
+        payload_truncated,
     }
 }
 
@@ -997,7 +1020,7 @@ fn spawn_v5_loop(
                     match event {
                         Ok(V5Event::Incoming(V5Packet::Publish(p))) => {
                             let topic = String::from_utf8_lossy(&p.topic).to_string();
-                            let record = v5_record(engine.next_seq(), &id, &p);
+                            let record = v5_record(&engine, engine.next_seq(), &id, &p);
                             engine.log(
                                 &id,
                                 LogLevel::Info,
@@ -1158,12 +1181,14 @@ fn spawn_v4_loop(
                 event = eventloop.poll() => {
                     match event {
                         Ok(rv4::Event::Incoming(rv4::Packet::Publish(p))) => {
+                            let (payload, payload_truncated) =
+                                crate::model::retained_payload(&p.payload);
                             let record = MqttRecord {
                                 seq: engine.next_seq(),
-                                connection_id: id.clone(),
+                                connection_id: engine.intern(&id),
                                 direction: Direction::Received,
-                                topic: String::from_utf8_lossy(&p.topic).to_string(),
+                                topic: engine.intern(&String::from_utf8_lossy(&p.topic)),
-                                payload: String::from_utf8_lossy(&p.payload).to_string(),
+                                payload,
                                 qos: match p.qos {
                                     rv4::QoS::AtMostOnce => 0,
                                     rv4::QoS::AtLeastOnce => 1,
@@ -1177,6 +1202,7 @@ fn spawn_v4_loop(
                                 correlation_data: None,
                                 message_expiry_interval: None,
                                 subscription_identifier: None,
+                                payload_truncated,
                             };
                             engine.log(
                                 &id, LogLevel::Info, "publish_received",

gpui-app/src/ui/app.rs

@@ -2,6 +2,7 @@
 
 use std::collections::{HashMap, VecDeque};
 use std::path::PathBuf;
+use std::sync::Arc;
 
 use gpui_kit::component::button::{Button, ButtonVariants as _, ButtonVariant};
 use gpui_kit::component::scroll::ScrollableElement as _;
@@ -24,7 +25,7 @@ use crate::aliyun::AliyunPreset;
 use crate::model::{
     AppSettings, ConnectionConfig, ConnectionStatus, Direction, GlobalVariable, LogEntry,
     LogLevel, MqttRecord, PublishParams, PublishPreset, SubscribeOptions, Subscription,
-    ThemeModePref,
+    ThemeModePref, retained_payload,
 };
 use crate::mqtt::{EngineEvent, MqttEngine};
 use crate::store::Storage;
@@ -83,7 +84,8 @@ pub struct MqttXApp {
     // ── 运行时状态 ──
     pub statuses: HashMap<String, ConnectionStatus>,
     pub errors: HashMap<String, String>,
-    pub messages: HashMap<String, Vec<MqttRecord>>,
+    /// 每连接消息环形缓冲：键为驻留后的连接 ID，容量受 max_messages 约束
+    pub messages: HashMap<Arc<str>, VecDeque<MqttRecord>>,
     pub logs: VecDeque<LogEntry>,
     pub seq: u64,
 
@@ -308,7 +310,7 @@ impl MqttXApp {
                 let id = record.connection_id.clone();
                 record.seq = self.next_seq();
                 self.push_message(id.clone(), record);
-                if let Some(view) = self.views.get(&id) {
+                if let Some(view) = self.views.get(id.as_ref()) {
                     view.update(cx, |_, cx| cx.notify());
                 }
             }
@@ -321,11 +323,12 @@ impl MqttXApp {
                 content_type,
                 user_properties,
             } => {
+                let (payload, payload_truncated) = retained_payload(&payload);
                 let record = MqttRecord {
                     seq: self.next_seq(),
-                    connection_id: connection_id.clone(),
+                    connection_id: Arc::from(connection_id.as_str()),
                     direction: Direction::Published,
-                    topic,
+                    topic: Arc::from(topic.as_str()),
                     payload,
                     qos,
                     retain,
@@ -336,9 +339,10 @@ impl MqttXApp {
                     correlation_data: None,
                     message_expiry_interval: None,
                     subscription_identifier: None,
+                    payload_truncated,
                 };
-                self.push_message(connection_id.clone(), record);
+                self.push_message(Arc::from(connection_id.as_str()), record);
-                if let Some(view) = self.views.get(&connection_id) {
+                if let Some(view) = self.views.get(connection_id.as_str()) {
                     view.update(cx, |_, cx| cx.notify());
                 }
             }
@@ -400,14 +404,17 @@ impl MqttXApp {
         self.seq
     }
 
-    fn push_message(&mut self, connection_id: String, record: MqttRecord) {
+    fn push_message(&mut self, connection_id: Arc<str>, record: MqttRecord) {
         let cap = self.settings.max_messages.max(100);
-        let list = self.messages.entry(connection_id).or_default();
+        let list = self
-        if list.len() >= cap {
+            .messages
-            let drop_n = list.len() - cap + 1;
+            .entry(connection_id)
-            list.drain(0..drop_n);
+            .or_insert_with(|| VecDeque::with_capacity(cap));
+        // 环形缓冲：满则丢最旧，均摊 O(1)（原来的 Vec::drain 是 O(n) 搬移）
+        while list.len() >= cap {
+            list.pop_front();
         }
-        list.push(record);
+        list.push_back(record);
     }
 
     fn push_log(&mut self, entry: LogEntry) {

gpui-app/src/ui/connection_view.rs

@@ -24,8 +24,9 @@ use gpui_kit::{
 use gpui_kit::component::IndexPath;
 
 use crate::model::{
-    render_template, render_will_templates, ConnectionConfig, ConnectionStatus, Direction, GlobalVariable,
+    render_template, render_will_templates, ConnectionConfig, ConnectionStatus, Direction,
-    LogLevel, MqttRecord, PayloadFormat, PublishParams, SubscribeOptions, Subscription,
+    GlobalVariable, LogLevel, MqttRecord, PayloadFormat, PublishParams, SubscribeOptions,
+    Subscription, MAX_PAYLOAD_RETAIN,
 };
 use crate::mqtt::MqttEngine;
 use crate::ui::app::MqttXApp;
@@ -1285,7 +1286,7 @@ impl ConnectionView {
                     .font_family(mono.clone())
                     // 匹配到订阅色时主题文字着色，否则保持默认前景色
                     .when_some(sub_color, |t, hue| t.text_color(hue_color(hue)))
-                    .child(record.topic.clone()),
+                    .child(record.topic.to_string()),
             )
             .child(
                 div()
@@ -1315,6 +1316,17 @@ impl ConnectionView {
                         .child("retain"),
                 )
             })
+            .when(record.payload_truncated, |t| {
+                t.child(
+                    div()
+                        .text_xs()
+                        .px_1()
+                        .rounded_sm()
+                        .bg(cx.theme().warning.alpha(0.15))
+                        .text_color(cx.theme().warning)
+                        .child(format!("已截断 {}KB", MAX_PAYLOAD_RETAIN / 1024)),
+                )
+            })
             .child(
                 div()
                     .text_xs()
@@ -1385,7 +1397,7 @@ impl ConnectionView {
                         .tooltip("复制主题")
                         .on_click(move |_, window, cx| {
                             cx.stop_propagation();
-                            cx.write_to_clipboard(ClipboardItem::new_string(copy_topic.clone()));
+                            cx.write_to_clipboard(ClipboardItem::new_string(copy_topic.to_string()));
                             window.push_notification(Notification::success("已复制主题"), cx);
                         }),
                 )
@@ -1398,7 +1410,7 @@ impl ConnectionView {
                         .tooltip("复制负载原文")
                         .on_click(move |_, window, cx| {
                             cx.stop_propagation();
-                            cx.write_to_clipboard(ClipboardItem::new_string(copy_payload.clone()));
+                            cx.write_to_clipboard(ClipboardItem::new_string(copy_payload.to_string()));
                             window.push_notification(Notification::success("已复制负载"), cx);
                         }),
                 )
@@ -1533,7 +1545,7 @@ impl ConnectionView {
                     .collect();
                 let mut matched = 0usize;
                 let mut records: Vec<MqttRecord> = Vec::new();
-                if let Some(m) = app.messages.get(&conn) {
+                if let Some(m) = app.messages.get(conn.as_str()) {
                     for r in m.iter().rev() {
                         // 方向 / 搜索 / 订阅过滤叠加，过滤后再截断保证最新消息优先
                         let dir_ok = match dir {
@@ -1742,7 +1754,7 @@ impl ConnectionView {
                                                 view.detail_format = DetailFormat::Auto;
                                                 if let Some(app) = view.app.upgrade() {
                                                     app.update(cx, |app, cx| {
-                                                        app.messages.remove(&clear_id);
+                                                        app.messages.remove(clear_id.as_str());
                                                         cx.notify();
                                                     });
                                                 }

gpui-app/tests/engine_live.rs

@@ -108,7 +108,7 @@ fn wait_subscribe(rx: &smol::channel::Receiver<EngineEvent>, topic: &str) -> boo
 
 fn wait_message(rx: &smol::channel::Receiver<EngineEvent>, topic: &str) -> bool {
     pump(rx, Duration::from_secs(10), |ev| match ev {
-        EngineEvent::Message(m) => m.topic == topic && m.payload == "hello-gpui",
+        EngineEvent::Message(m) => &*m.topic == topic && &*m.payload == "hello-gpui",
         _ => false,
     })
 }