[ISSUE #1281]⚡️Optimize name server DefaultRequestProcessor#delete_topic_in_name_srv

mxsm · mxsm · commit d4072c58bdcf · 2024-11-24T22:40:35.000+08:00
diff --git a/rocketmq-namesrv/src/processor/default_request_processor.rs b/rocketmq-namesrv/src/processor/default_request_processor.rs
@@ -349,10 +349,8 @@ impl DefaultRequestProcessor {
         let request_header = request
             .decode_command_custom_header::<DeleteTopicFromNamesrvRequestHeader>()
             .expect("decode DeleteTopicFromNamesrvRequestHeader failed");
-        self.route_info_manager.delete_topic(
-            request_header.topic.as_str(),
-            request_header.cluster_name.clone(),
-        );
+        self.route_info_manager
+            .delete_topic(request_header.topic, request_header.cluster_name);
         RemotingCommand::create_response_command()
     }
 
diff --git a/rocketmq-namesrv/src/route/route_info_manager.rs b/rocketmq-namesrv/src/route/route_info_manager.rs
@@ -736,29 +736,30 @@ impl RouteInfoManager {
 
     pub(crate) fn delete_topic(
         &mut self,
-        topic: impl Into<String>,
-        cluster_name: Option<impl Into<String>>,
+        topic: CheetahString,
+        cluster_name: Option<CheetahString>,
     ) {
-        let topic_inner = topic.into();
-        if cluster_name.is_some() {
-            let cluster_name_inner = cluster_name.map(|s| s.into()).unwrap();
-            let broker_names = self.cluster_addr_table.get(cluster_name_inner.as_str());
+        let lock = self.lock.write();
+        if cluster_name.as_ref().is_some_and(|inner| !inner.is_empty()) {
+            let cluster_name_inner = cluster_name.unwrap();
+            let broker_names = self.cluster_addr_table.get(&cluster_name_inner);
             if broker_names.is_none() || broker_names.unwrap().is_empty() {
                 return;
             }
-            if let Some(queue_data_map) = self.topic_queue_table.get_mut(topic_inner.as_str()) {
+            if let Some(queue_data_map) = self.topic_queue_table.mut_from_ref().get_mut(&topic) {
                 for broker_name in broker_names.unwrap() {
                     if let Some(remove_qd) = queue_data_map.remove(broker_name) {
                         info!(
                             "deleteTopic, remove one broker's topic {} {} {:?}",
-                            broker_name, &topic_inner, remove_qd
+                            broker_name, &topic, remove_qd
                         )
                     }
                 }
             }
         } else {
-            self.topic_queue_table.remove(topic_inner.as_str());
+            self.topic_queue_table.mut_from_ref().remove(&topic);
         }
+        drop(lock)
     }
 
     pub(crate) fn register_topic(&mut self, topic: CheetahString, queue_data_vec: Vec<QueueData>) {

Original file line number	Diff line number	Diff line change
`@@ -736,29 +736,30 @@ impl RouteInfoManager {`
`736`	`736`
`737`	`737`	`pub(crate) fn delete_topic(`
`738`	`738`	`&mut self,`
`739`		`- topic: impl Into<String>,`
`740`		`- cluster_name: Option<impl Into<String>>,`
	`739`	`+ topic: CheetahString,`
	`740`	`+ cluster_name: Option<CheetahString>,`
`741`	`741`	`) {`
`742`		`- let topic_inner = topic.into();`
`743`		`- if cluster_name.is_some() {`
`744`		`- let cluster_name_inner = cluster_name.map(\|s\| s.into()).unwrap();`
`745`		`- let broker_names = self.cluster_addr_table.get(cluster_name_inner.as_str());`
	`742`	`+ let lock = self.lock.write();`
	`743`	`+ if cluster_name.as_ref().is_some_and(\|inner\| !inner.is_empty()) {`
	`744`	`+ let cluster_name_inner = cluster_name.unwrap();`
	`745`	`+ let broker_names = self.cluster_addr_table.get(&cluster_name_inner);`
`746`	`746`	`if broker_names.is_none() \|\| broker_names.unwrap().is_empty() {`
`747`	`747`	`return;`
`748`	`748`	`}`
`749`		`- if let Some(queue_data_map) = self.topic_queue_table.get_mut(topic_inner.as_str()) {`
	`749`	`+ if let Some(queue_data_map) = self.topic_queue_table.mut_from_ref().get_mut(&topic) {`
`750`	`750`	`for broker_name in broker_names.unwrap() {`
`751`	`751`	`if let Some(remove_qd) = queue_data_map.remove(broker_name) {`
`752`	`752`	`info!(`
`753`	`753`	`"deleteTopic, remove one broker's topic {} {} {:?}",`
`754`		`- broker_name, &topic_inner, remove_qd`
	`754`	`+ broker_name, &topic, remove_qd`
`755`	`755`	`)`
`756`	`756`	`}`
`757`	`757`	`}`
`758`	`758`	`}`
`759`	`759`	`} else {`
`760`		`- self.topic_queue_table.remove(topic_inner.as_str());`
	`760`	`+ self.topic_queue_table.mut_from_ref().remove(&topic);`
`761`	`761`	`}`
	`762`	`+ drop(lock)`
`762`	`763`	`}`
`763`	`764`
`764`	`765`	`pub(crate) fn register_topic(&mut self, topic: CheetahString, queue_data_vec: Vec<QueueData>) {`