[ISSUE #479]🔥Optimize broker register and client scan_available_name_srv🎨

rocketmq-broker/Cargo.toml

@@ -55,7 +55,7 @@ rand = "0.8.5"
 dirs.workspace = true
 
 local-ip-address = "0.6.1"
-log = "0.4.21"
+dns-lookup = "2.0"
 
 [dev-dependencies]
 mockall = "0.12.1"

rocketmq-broker/src/broker_runtime.rs

@@ -450,6 +450,30 @@ impl BrokerRuntime {
         if self.broker_config.enable_controller_mode {
             self.update_master_haserver_addr_periodically = true;
         }
+
+        if let Some(ref namesrv_address) = self.broker_config.namesrv_addr.clone() {
+            self.update_namesrv_addr();
+            info!(
+                "Set user specified name server address: {}",
+                namesrv_address
+            );
+            let mut broker_runtime = self.clone();
+            self.broker_runtime
+                .as_ref()
+                .unwrap()
+                .get_handle()
+                .spawn(async move {
+                    tokio::time::sleep(Duration::from_secs(10)).await;
+                    loop {
+                        let current_execution_time = tokio::time::Instant::now();
+                        broker_runtime.update_namesrv_addr();
+                        let next_execution_time = current_execution_time + Duration::from_secs(60);
+                        let delay = next_execution_time
+                            .saturating_duration_since(tokio::time::Instant::now());
+                        tokio::time::sleep(delay).await;
+                    }
+                });
+        }
     }
 
     fn initial_transaction(&mut self) {}
@@ -475,6 +499,18 @@ impl BrokerRuntime {
         tokio::spawn(async move { server.run(request_processor).await });
     }
 
+    fn update_namesrv_addr(&mut self) {
+        if self.broker_config.fetch_name_srv_addr_by_dns_lookup {
+            if let Some(namesrv_addr) = &self.broker_config.namesrv_addr {
+                self.broker_out_api
+                    .update_name_server_address_list_by_dns_lookup(namesrv_addr.clone());
+            }
+        } else if let Some(namesrv_addr) = &self.broker_config.namesrv_addr {
+            self.broker_out_api
+                .update_name_server_address_list(namesrv_addr.clone());
+        }
+    }
+
     pub async fn start(&mut self) {
         self.should_start_time.store(
             (get_current_millis() as i64 + self.message_store_config.disappear_time_after_start)
@@ -572,7 +608,7 @@ impl BrokerRuntime {
                     tokio::time::sleep(delay).await;
                 }
             });
-        log::info!(
+        info!(
             "Rocketmq Broker({} ----Rust) start success",
             self.broker_config.broker_identity.broker_name
         );

rocketmq-broker/src/out_api/broker_outer_api.rs

@@ -16,6 +16,7 @@
  */
 use std::sync::Arc;
 
+use dns_lookup::lookup_host;
 use rocketmq_common::common::broker::broker_config::BrokerIdentity;
 use rocketmq_common::common::config::TopicConfig;
 use rocketmq_common::utils::crc32_utils;
@@ -34,6 +35,8 @@ use rocketmq_remoting::protocol::RemotingSerializable;
 use rocketmq_remoting::remoting::RemotingService;
 use rocketmq_remoting::runtime::config::client_config::TokioClientConfig;
 use rocketmq_remoting::runtime::RPCHook;
+use tracing::error;
+use tracing::info;
 
 #[derive(Clone)]
 pub struct BrokerOuterAPI {
@@ -99,6 +102,12 @@ impl BrokerOuterAPI {
             .update_name_server_address_list(addr_vec)
     }
 
+    pub fn update_name_server_address_list_by_dns_lookup(&self, domain: String) {
+        let address_list = dns_lookup_address_by_domain(domain.as_str());
+        self.remoting_client
+            .update_name_server_address_list(address_list);
+    }
+
     pub async fn register_broker_all(
         &self,
         cluster_name: String,
@@ -225,3 +234,62 @@ impl BrokerOuterAPI {
 
     pub fn refresh_metadata(&self) {}
 }
+
+fn dns_lookup_address_by_domain(domain: &str) -> Vec<String> {
+    let mut address_list = Vec::new();
+    // Ensure logging is initialized
+
+    match domain.find(':') {
+        Some(index) => {
+            let (domain_str, port_str) = domain.split_at(index);
+            match lookup_host(domain_str) {
+                Ok(addresses) => {
+                    for address in addresses {
+                        address_list.push(format!("{}{}", address, port_str));
+                    }
+                    info!(
+                        "DNS lookup address by domain success, domain={}, result={:?}",
+                        domain, address_list
+                    );
+                }
+                Err(e) => {
+                    error!(
+                        "DNS lookup address by domain error, domain={}, error={}",
+                        domain, e
+                    );
+                }
+            }
+        }
+        None => {
+            error!("Invalid domain format, missing port: {}", domain);
+        }
+    }
+
+    address_list
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    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()));
+    }
+
+    #[test]
+    fn dns_lookup_address_by_domain_handles_invalid_domain() {
+        let domain = "invalid_domain";
+        let addresses = dns_lookup_address_by_domain(domain);
+        assert!(addresses.is_empty());
+    }
+
+    #[test]
+    fn dns_lookup_address_by_domain_handles_domain_without_port() {
+        let domain = "localhost";
+        let addresses = dns_lookup_address_by_domain(domain);
+        assert!(addresses.is_empty());
+    }
+}

rocketmq-common/src/common/broker/broker_config.rs

@@ -20,6 +20,7 @@ use serde::Deserialize;
 
 use crate::common::constant::PermName;
 use crate::common::mix_all;
+use crate::common::mix_all::NAMESRV_ADDR_PROPERTY;
 use crate::common::server::config::ServerConfig;
 use crate::common::topic::TopicValidator;
 
@@ -34,6 +35,8 @@ lazy_static! {
             None
         }
     };
+    pub static ref NAMESRV_ADDR: Option<String> =
+        std::env::var(NAMESRV_ADDR_PROPERTY).map_or(Some("127.0.0.1:9876".to_string()), Some);
 }
 
 #[derive(Debug, Default, Deserialize)]
@@ -137,6 +140,8 @@ pub struct BrokerConfig {
     pub force_register: bool,
     pub register_name_server_period: u64,
     pub skip_pre_online: bool,
+    pub namesrv_addr: Option<String>,
+    pub fetch_name_srv_addr_by_dns_lookup: bool,
 }
 
 impl Default for BrokerConfig {
@@ -188,6 +193,8 @@ impl Default for BrokerConfig {
             force_register: true,
             register_name_server_period: 1000 * 30,
             skip_pre_online: false,
+            namesrv_addr: NAMESRV_ADDR.clone(),
+            fetch_name_srv_addr_by_dns_lookup: false,
         }
     }
 }

rocketmq-remoting/Cargo.toml

@@ -14,6 +14,7 @@ readme = "README.md"
 [dependencies]
 rocketmq-common = { workspace = true }
 rocketmq-macros = { workspace = true }
+rocketmq-runtime = { workspace = true }
 
 anyhow.workspace = true
 bytes.workspace = true

rocketmq-remoting/src/clients.rs

@@ -91,7 +91,7 @@ pub trait RemotingClient: RemotingService {
         addr: String,
         request: RemotingCommand,
         timeout_millis: u64,
-    ) -> RemotingCommand;
+    ) -> Result<RemotingCommand, RemotingError>;
 
     async fn invoke_async(
         &mut self,

rocketmq-remoting/src/clients/client.rs

@@ -44,7 +44,7 @@ impl Client {
     /// A new `Client` instance wrapped in a `Result`. Returns an error if the connection fails.
     pub async fn connect<T: tokio::net::ToSocketAddrs>(addr: T) -> anyhow::Result<Client> {
         let tcp_stream = tokio::net::TcpStream::connect(addr).await?;
-        let socket_addr = tcp_stream.peer_addr().unwrap();
+        let socket_addr = tcp_stream.peer_addr()?;
         Ok(Client {
             connection: Connection::new(tcp_stream, socket_addr),
         })