perf(broker): ordered batched fan-out worker for non-persistent publish

rust/Cargo.toml

@@ -6,6 +6,7 @@ members = [
     "storage/resources",
     "storage/managed-ledger",
     "storage/managed-ledger-rocksdb",
+    "proto",
 ]
 resolver = "2"
 
@@ -21,6 +22,8 @@ thiserror = "1.0"
 tokio = { version = "1.35", features = ["full"] }
 anyhow = "1.0"
 log = "0.4"
+tikv-jemallocator = { version = "0.6", features = ["unprefixed_malloc_on_supported_platforms"] }
+tikv-jemalloc-sys = { version = "0.6", features = ["profiling", "unprefixed_malloc_on_supported_platforms"] }
 
 [package]
 name = "pulsar-lite"
@@ -45,6 +48,7 @@ pulsar-lite-storage-metadata = { path = "storage/metadata" }
 pulsar-lite-storage-resources = { path ="storage/resources" }
 pulsar-lite-storage-managed-ledger = { path = "storage/managed-ledger" }
 pulsar-lite-storage-managed-ledger-rocksdb = { path = "storage/managed-ledger-rocksdb", optional = true }
+pulsar-lite-proto = { path  =  "proto" }
 
 # 异步运行时
 tokio = { version = "1.35", features = ["full"] }
@@ -82,6 +86,10 @@ uuid = { version = "1.6", features = ["v4"] }
 tokio-stream = "0.1"
 futures = "0.3"
 
+# 内存分配器：jemalloc（缓解 glibc arena 内存滞留导致的 RSS 高水位）
+tikv-jemallocator.workspace = true
+tikv-jemalloc-sys.workspace = true
+
 [build-dependencies]
 prost-build = "0.13"
 

rust/proto/Cargo.toml

@@ -0,0 +1,14 @@
+[package]
+name = "pulsar-lite-proto"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+# Protobuf
+prost = "0.13"
+prost-types = "0.13"
+bytes = "1.5"
+tokio-util = { version = "0.7", features = ["codec"] }
+
+[build-dependencies]
+prost-build = "0.13"

rust/build.rs → rust/proto/build.rs

@@ -4,10 +4,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
     prost_build::Config::new()
         .out_dir(&out_dir)
         .protoc_arg("--experimental_allow_proto3_optional")
-        .compile_protos(
-            &["proto/PulsarApi.proto", "proto/MLDataFormats.proto"],
-            &["proto"],
-        )?;
+        .compile_protos(&["PulsarApi.proto", "MLDataFormats.proto"], &["proto"])?;
 
     Ok(())
 }

rust/src/protocol/codec.rs → rust/proto/src/codec.rs

@@ -21,7 +21,6 @@ pub mod proto {
 
 // Import ServerCommand from command module
 use super::command::ServerCommand;
-use crate::broker::service::PendingMessage;
 
 /// Magic number for checksum verification (0x0e01)
 const MAGIC_NUMBER: u16 = 0x0e01;
@@ -305,31 +304,9 @@ impl Encoder<ServerCommand> for PulsarFrameCodec {
     }
 }
 
-impl Encoder<(u64, PendingMessage)> for PulsarFrameCodec {
-    type Error = io::Error;
-
-    fn encode(
-        &mut self,
-        item: (u64, PendingMessage),
-        dst: &mut BytesMut,
-    ) -> Result<(), Self::Error> {
-        let (consumer_id, msg) = item;
-        self.encode_message(
-            consumer_id,
-            msg.message_id.ledger,
-            msg.message_id.entry,
-            msg.message_id.partition,
-            &msg.metadata,
-            &msg.payload,
-            msg.redelivery_count,
-            dst,
-        )
-    }
-}
-
 impl PulsarFrameCodec {
     /// Encode a Message command with payload
-    fn encode_message(
+    pub fn encode_message(
         &self,
         consumer_id: u64,
         ledger_id: u64,
@@ -360,9 +337,9 @@ impl PulsarFrameCodec {
 
 #[cfg(test)]
 mod tests {
-    use super::proto::pulsar::{BaseCommand, CompressionType, KeyValue, MessageMetadata};
     use super::*;
-    use crate::protocol::command::ServerCommand;
+    use crate::codec::proto::pulsar::{BaseCommand, CompressionType, KeyValue, MessageMetadata};
+    use crate::command::ServerCommand;
     use bytes::Bytes;
     use prost::Message;
 

rust/src/broker/broker_service.rs

@@ -8,7 +8,7 @@ use crate::broker::service::topic::{
     PartitionedTopic, PartitionedTopicStats, SharedPartitionedTopic, Topic, TopicPublishRate,
     TopicStats,
 };
-use crate::storage::Storage;
+use pulsar_lite_storage::Storage;
 use std::collections::HashMap;
 use std::sync::Arc;
 use tokio::sync::{Mutex, RwLock};

rust/src/broker/dispatcher/enums.rs

@@ -8,7 +8,7 @@ use super::{ExclusiveDispatcher, FailoverDispatcher, KeySharedDispatcher, Shared
 use crate::broker::dispatcher::redelivery_controller::RedeliveryEntry;
 use crate::broker::service::topic::{KeySharedPolicy, SubscriptionType};
 use crate::broker::service::{Consumer, SharedStorage};
-use crate::storage::{ManagedLedgerPosition, MessageId};
+use pulsar_lite_storage_managed_ledger::{ManagedLedgerPosition, MessageId};
 use std::sync::Arc;
 
 /// Dispatcher enum - holds the concrete dispatcher implementation

rust/src/broker/dispatcher/exclusive.rs

@@ -10,7 +10,7 @@ use super::read_position::{commit_read_position, next_unacked_candidate};
 use crate::broker::dispatcher::Dispatcher;
 use crate::broker::service::topic::SubscriptionType;
 use crate::broker::service::{Consumer, SharedStorage};
-use crate::storage::ManagedLedgerPosition;
+use pulsar_lite_storage_managed_ledger::ManagedLedgerPosition;
 use std::sync::atomic::{AtomicU32, Ordering};
 use std::sync::{Arc, RwLock};
 

rust/src/broker/dispatcher/failover.rs

@@ -10,7 +10,7 @@ use super::rewind_read_position;
 use crate::broker::dispatcher::Dispatcher;
 use crate::broker::service::topic::SubscriptionType;
 use crate::broker::service::{Consumer, SharedStorage};
-use crate::storage::ManagedLedgerPosition;
+use pulsar_lite_storage_managed_ledger::ManagedLedgerPosition;
 use std::sync::atomic::{AtomicU32, Ordering};
 use std::sync::{Arc, RwLock};
 
@@ -325,7 +325,7 @@ impl Dispatcher for FailoverDispatcher {
 mod tests {
     use super::*;
     use crate::broker::service::topic::Subscription;
-    use crate::storage::Storage;
+    use pulsar_lite_storage::Storage;
     use std::path::Path;
     use tokio::sync::{mpsc, Mutex, RwLock};