[ISSUE #1173]🔥Optimize use CheetahString replace Std String🎨 (#1174)

Author: mxsm

Cargo.lock

@@ -81,4 +81,4 @@ uuid = { version = "1.11.0", features = ["v4", # Lets you generate random UUIDs
 
 futures = "0.3"
 
-cheetah-string = { version = "0.1.4", features = ["serde", "bytes"] }
+cheetah-string = { version = "0.1.5", features = ["serde", "bytes"] }

Cargo.toml

@@ -216,7 +216,7 @@ impl BrokerRuntime {
         );
         broker_member_group.broker_addrs.insert(
             broker_config.broker_identity.broker_id,
-            broker_config.get_broker_addr(),
+            broker_config.get_broker_addr().into(),
         );
         Self {
             broker_config: broker_config.clone(),

rocketmq-broker/src/broker_runtime.rs

@@ -14,22 +14,27 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-
+use cheetah_string::CheetahString;
 use rocketmq_common::TimeUtils::get_current_millis;
 use rocketmq_remoting::net::channel::Channel;
 use rocketmq_remoting::protocol::LanguageCode;
 
 #[derive(Debug, Clone, Hash, PartialEq)]
 pub struct ClientChannelInfo {
     channel: Channel,
-    client_id: String,
+    client_id: CheetahString,
     language: LanguageCode,
     version: i32,
     last_update_timestamp: u64,
 }
 
 impl ClientChannelInfo {
-    pub fn new(channel: Channel, client_id: String, language: LanguageCode, version: i32) -> Self {
+    pub fn new(
+        channel: Channel,
+        client_id: CheetahString,
+        language: LanguageCode,
+        version: i32,
+    ) -> Self {
         Self {
             channel,
             client_id,
@@ -39,7 +44,7 @@ impl ClientChannelInfo {
         }
     }
 
-    pub fn client_id(&self) -> &String {
+    pub fn client_id(&self) -> &CheetahString {
         &self.client_id
     }
 
@@ -55,8 +60,8 @@ impl ClientChannelInfo {
         self.last_update_timestamp
     }
 
-    pub fn set_client_id(&mut self, client_id: String) {
-        self.client_id = client_id;
+    pub fn set_client_id(&mut self, client_id: impl Into<CheetahString>) {
+        self.client_id = client_id.into();
     }
 
     pub fn set_language(&mut self, language: LanguageCode) {

rocketmq-broker/src/client/client_channel_info.rs

@@ -18,6 +18,7 @@ use std::collections::HashMap;
 use std::collections::HashSet;
 use std::sync::Arc;
 
+use cheetah_string::CheetahString;
 use parking_lot::Mutex;
 use parking_lot::RwLock;
 use rocketmq_common::common::consumer::consume_from_where::ConsumeFromWhere;
@@ -34,8 +35,8 @@ use crate::client::client_channel_info::ClientChannelInfo;
 
 #[derive(Debug, Clone)]
 pub struct ConsumerGroupInfo {
-    group_name: String,
-    subscription_table: Arc<RwLock<HashMap<String, SubscriptionData>>>,
+    group_name: CheetahString,
+    subscription_table: Arc<RwLock<HashMap<CheetahString, SubscriptionData>>>,
     channel_info_table: Arc<RwLock<HashMap<Channel, ClientChannelInfo>>>,
     consume_type: Arc<RwLock<ConsumeType>>,
     message_model: Arc<RwLock<MessageModel>>,
@@ -45,13 +46,13 @@ pub struct ConsumerGroupInfo {
 
 impl ConsumerGroupInfo {
     pub fn new(
-        group_name: String,
+        group_name: impl Into<CheetahString>,
         consume_type: ConsumeType,
         message_model: MessageModel,
         consume_from_where: ConsumeFromWhere,
     ) -> Self {
         ConsumerGroupInfo {
-            group_name,
+            group_name: group_name.into(),
             subscription_table: Arc::new(RwLock::new(HashMap::new())),
             channel_info_table: Arc::new(RwLock::new(HashMap::new())),
             consume_type: Arc::new(RwLock::new(consume_type)),
@@ -61,9 +62,9 @@ impl ConsumerGroupInfo {
         }
     }
 
-    pub fn with_group_name(group_name: String) -> Self {
+    pub fn with_group_name(group_name: impl Into<CheetahString>) -> Self {
         ConsumerGroupInfo {
-            group_name,
+            group_name: group_name.into(),
             subscription_table: Arc::new(RwLock::new(HashMap::new())),
             channel_info_table: Arc::new(RwLock::new(HashMap::new())),
             consume_type: Arc::new(RwLock::new(ConsumeType::ConsumePassively)),
@@ -83,7 +84,7 @@ impl ConsumerGroupInfo {
         None
     }
 
-    pub fn get_subscription_table(&self) -> Arc<RwLock<HashMap<String, SubscriptionData>>> {
+    pub fn get_subscription_table(&self) -> Arc<RwLock<HashMap<CheetahString, SubscriptionData>>> {
         Arc::clone(&self.subscription_table)
     }
 
@@ -101,7 +102,7 @@ impl ConsumerGroupInfo {
         channel_info_table.keys().cloned().collect()
     }
 
-    pub fn get_all_client_ids(&self) -> Vec<String> {
+    pub fn get_all_client_ids(&self) -> Vec<CheetahString> {
         let channel_info_table = self.channel_info_table.read();
         channel_info_table
             .values()
@@ -195,7 +196,7 @@ impl ConsumerGroupInfo {
 
     pub fn update_subscription(&self, sub_list: &HashSet<SubscriptionData>) -> bool {
         let mut updated = false;
-        let mut topic_set: HashSet<String> = HashSet::new();
+        let mut topic_set = HashSet::new();
 
         let mut subscription_table = self.subscription_table.write();
         for sub in sub_list.iter() {
@@ -238,7 +239,7 @@ impl ConsumerGroupInfo {
         updated
     }
 
-    pub fn get_subscribe_topics(&self) -> HashSet<String> {
+    pub fn get_subscribe_topics(&self) -> HashSet<CheetahString> {
         let subscription_table = self.subscription_table.read();
         subscription_table.keys().cloned().collect()
     }
@@ -266,7 +267,7 @@ impl ConsumerGroupInfo {
         *message_model_lock = message_model;
     }
 
-    pub fn get_group_name(&self) -> &String {
+    pub fn get_group_name(&self) -> &CheetahString {
         &self.group_name
     }
 
@@ -395,8 +396,8 @@ mod tests {
 
         let mut sub_list = HashSet::new();
         let subscription_data = SubscriptionData {
-            topic: "topic".to_string(),
-            sub_string: "sub_string".to_string(),
+            topic: "topic".into(),
+            sub_string: "sub_string".into(),
             ..Default::default()
         };
         sub_list.insert(subscription_data);

rocketmq-broker/src/client/consumer_group_info.rs

@@ -146,7 +146,7 @@ impl ConsumerManager {
         consumer_group_info
             .get_subscription_table()
             .write()
-            .insert(topic.to_string(), subscription_data.clone());
+            .insert(topic.into(), subscription_data.clone());
     }
 
     pub fn compensate_basic_consumer_info(

rocketmq-broker/src/client/manager/consumer_manager.rs

@@ -17,6 +17,7 @@
 
 use std::collections::HashMap;
 
+use cheetah_string::CheetahString;
 use rocketmq_common::TimeUtils::get_current_millis;
 use rocketmq_remoting::net::channel::Channel;
 use rocketmq_remoting::runtime::connection_handler_context::ConnectionHandlerContext;
@@ -26,9 +27,10 @@ use crate::client::client_channel_info::ClientChannelInfo;
 
 #[derive(Default)]
 pub struct ProducerManager {
-    group_channel_table:
-        parking_lot::Mutex<HashMap<String /* group name */, HashMap<Channel, ClientChannelInfo>>>,
-    client_channel_table: parking_lot::Mutex<HashMap<String, Channel /* client ip:port */>>,
+    group_channel_table: parking_lot::Mutex<
+        HashMap<CheetahString /* group name */, HashMap<Channel, ClientChannelInfo>>,
+    >,
+    client_channel_table: parking_lot::Mutex<HashMap<CheetahString, Channel /* client ip:port */>>,
 }
 
 impl ProducerManager {
@@ -43,7 +45,7 @@ impl ProducerManager {
 impl ProducerManager {
     pub fn group_online(&self, group: String) -> bool {
         let binding = self.group_channel_table.lock();
-        let channels = binding.get(&group);
+        let channels = binding.get(group.as_str());
         if channels.is_none() {
             return false;
         }
@@ -82,10 +84,14 @@ impl ProducerManager {
     }
 
     #[allow(clippy::mutable_key_type)]
-    pub fn register_producer(&self, group: &str, client_channel_info: &ClientChannelInfo) {
+    pub fn register_producer(
+        &self,
+        group: &CheetahString,
+        client_channel_info: &ClientChannelInfo,
+    ) {
         let mut group_channel_table = self.group_channel_table.lock();
 
-        let key = group.to_string();
+        let key = group.clone();
         let channel_table = group_channel_table.entry(key).or_default();
 
         if let Some(client_channel_info_found) =

rocketmq-broker/src/client/manager/producer_manager.rs

@@ -17,6 +17,7 @@
 use std::collections::HashMap;
 use std::sync::Arc;
 
+use cheetah_string::CheetahString;
 use rocketmq_common::common::config::TopicConfig;
 use rocketmq_common::common::message::message_ext::MessageExt;
 use rocketmq_store::base::message_result::PutMessageResult;
@@ -25,11 +26,13 @@ use rocketmq_store::hook::put_message_hook::PutMessageHook;
 use crate::util::hook_utils::HookUtils;
 
 pub struct BatchCheckBeforePutMessageHook {
-    topic_config_table: Arc<parking_lot::Mutex<HashMap<String, TopicConfig>>>,
+    topic_config_table: Arc<parking_lot::Mutex<HashMap<CheetahString, TopicConfig>>>,
 }
 
 impl BatchCheckBeforePutMessageHook {
-    pub fn new(topic_config_table: Arc<parking_lot::Mutex<HashMap<String, TopicConfig>>>) -> Self {
+    pub fn new(
+        topic_config_table: Arc<parking_lot::Mutex<HashMap<CheetahString, TopicConfig>>>,
+    ) -> Self {
         Self { topic_config_table }
     }
 }

rocketmq-broker/src/hook/batch_check_before_put_message.rs

@@ -128,17 +128,17 @@ impl BrokerOuterAPI {
         self.remoting_client.start(wrapper).await;
     }
 
-    pub async fn update_name_server_address_list(&self, addrs: String) {
+    pub async fn update_name_server_address_list(&self, addrs: CheetahString) {
         let addr_vec = addrs
-            .split("';'")
-            .map(|s| s.to_string())
-            .collect::<Vec<String>>();
+            .split(";")
+            .map(CheetahString::from_slice)
+            .collect::<Vec<CheetahString>>();
         self.remoting_client
             .update_name_server_address_list(addr_vec)
             .await
     }
 
-    pub async fn update_name_server_address_list_by_dns_lookup(&self, domain: String) {
+    pub async fn update_name_server_address_list_by_dns_lookup(&self, domain: CheetahString) {
         let address_list = dns_lookup_address_by_domain(domain.as_str());
         self.remoting_client
             .update_name_server_address_list(address_list)
@@ -233,7 +233,7 @@ impl BrokerOuterAPI {
 
     async fn register_broker(
         &self,
-        namesrv_addr: String,
+        namesrv_addr: CheetahString,
         oneway: bool,
         timeout_mills: u64,
         request_header: RegisterBrokerRequestHeader,
@@ -338,7 +338,7 @@ impl BrokerOuterAPI {
 
     pub async fn lock_batch_mq_async(
         &self,
-        addr: String,
+        addr: CheetahString,
         request_body: bytes::Bytes,
         timeout_millis: u64,
     ) -> Result<HashSet<MessageQueue>> {
@@ -375,7 +375,7 @@ impl BrokerOuterAPI {
 
     pub async fn unlock_batch_mq_async(
         &self,
-        addr: String,
+        addr: CheetahString,
         request_body: bytes::Bytes,
         timeout_millis: u64,
     ) -> Result<()> {
@@ -409,7 +409,7 @@ impl BrokerOuterAPI {
     }
 }
 
-fn dns_lookup_address_by_domain(domain: &str) -> Vec<String> {
+fn dns_lookup_address_by_domain(domain: &str) -> Vec<CheetahString> {
     let mut address_list = Vec::new();
     // Ensure logging is initialized
 
@@ -419,7 +419,7 @@ fn dns_lookup_address_by_domain(domain: &str) -> Vec<String> {
             match lookup_host(domain_str) {
                 Ok(addresses) => {
                     for address in addresses {
-                        address_list.push(format!("{}{}", address, port_str));
+                        address_list.push(format!("{}{}", address, port_str).into());
                     }
                     info!(
                         "DNS lookup address by domain success, domain={}, result={:?}",
@@ -450,7 +450,7 @@ mod tests {
     fn dns_lookup_address_by_domain_returns_correct_addresses() {
         let domain = "localhost:8080";
         let addresses = dns_lookup_address_by_domain(domain);
-        assert!(addresses.contains(&"127.0.0.1:8080".to_string()));
+        assert!(addresses.contains(&"127.0.0.1:8080".into()));
     }
 
     #[test]

rocketmq-broker/src/out_api/broker_outer_api.rs

@@ -17,6 +17,7 @@
 
 use std::collections::HashMap;
 
+use cheetah_string::CheetahString;
 use rocketmq_common::common::mix_all;
 use rocketmq_common::common::mq_version::RocketMqVersion;
 use rocketmq_remoting::code::request_code::RequestCode;
@@ -98,7 +99,7 @@ impl BrokerConfigRequestHandler {
         Some(response)
     }
 
-    fn prepare_runtime_info(&self) -> HashMap<String, String> {
+    fn prepare_runtime_info(&self) -> HashMap<CheetahString, CheetahString> {
         let mut runtime_info = self.inner.default_message_store.get_runtime_info();
         self.inner
             .schedule_message_service
@@ -262,7 +263,7 @@ impl BrokerConfigRequestHandler {
             ),
         );
         let store_path_root_dir = &self.inner.message_store_config.store_path_root_dir;
-        let commit_log_dir = std::path::Path::new(&store_path_root_dir);
+        let commit_log_dir = std::path::Path::new(store_path_root_dir.as_str());
         if commit_log_dir.exists() {
             let disks = Disks::new_with_refreshed_list();
             let path_str = commit_log_dir.to_str().unwrap();
@@ -283,6 +284,9 @@ impl BrokerConfigRequestHandler {
             }
         }
         runtime_info
+            .into_iter()
+            .map(|(k, v)| (CheetahString::from_string(k), CheetahString::from_string(v)))
+            .collect()
     }
     fn is_special_service_running(&self) -> bool {
         true

rocketmq-broker/src/processor/admin_broker_processor/broker_config_request_handler.rs

@@ -80,7 +80,7 @@ impl ConsumerRequestHandler {
                     connection.set_client_id(info.client_id().clone());
                     connection.set_language(info.language());
                     connection.set_version(info.version());
-                    connection.set_client_addr(channel.remote_address().to_string());
+                    connection.set_client_addr(channel.remote_address().to_string().into());
                     body_data.get_connection_set().insert(connection);
                 }
                 let body = body_data.encode();
@@ -158,7 +158,7 @@ impl ConsumerRequestHandler {
             for i in 0..topic_config.unwrap().get_read_queue_nums() {
                 let mut mq = MessageQueue::new();
                 mq.set_topic(topic.to_string().into());
-                mq.set_broker_name(self.inner.broker_config.broker_name.clone().into());
+                mq.set_broker_name(self.inner.broker_config.broker_name.clone());
                 mq.set_queue_id(i as i32);
 
                 let mut offset_wrapper = OffsetWrapper::new();