[ISSUE #368]🚀Optimize send message logic (#369)

Author: mxsm

Cargo.toml

@@ -25,8 +25,8 @@ Unofficial Rust implementation of Apache RocketMQ
 """
 [workspace.dependencies]
 tokio = { version = "1.35", features = ["full"] }
-tokio-util = {version = "0.7.10",features = ["full"]}
-tokio-stream = {version = "0.1.14",features = ["full"]}
+tokio-util = { version = "0.7.10", features = ["full"] }
+tokio-stream = { version = "0.1.14", features = ["full"] }
 
 log = "0.4.0"
 env_logger = "0.11.2"
@@ -53,4 +53,6 @@ config = "0.14"
 
 parking_lot = "0.12"
 dirs = "5.0"
-trait-variant = "0.1.2"
+trait-variant = "0.1.2"
+
+once_cell = "1.19.0"

rocketmq-broker/src/broker_runtime.rs

@@ -174,6 +174,7 @@ impl BrokerRuntime {
         if let Some(message_store) = &mut self.message_store {
             message_store.shutdown()
         }
+
         if let Some(runtime) = self.broker_runtime.take() {
             runtime.shutdown();
         }

rocketmq-broker/src/processor/send_message_processor.rs

@@ -291,7 +291,7 @@ impl<MS: MessageStore + Send + Clone> SendMessageProcessor<MS> {
         message_ext.tags_code = MessageExtBrokerInner::tags_string2tags_code(
             &topic_config.topic_filter_type,
             message_ext.get_tags().unwrap_or("".to_string()).as_str(),
-        ) as i64;
+        );
 
         message_ext.message_ext_inner.born_timestamp = request_header.born_timestamp;
         message_ext.message_ext_inner.born_host = ctx.remoting_address();

rocketmq-common/Cargo.toml

@@ -49,4 +49,5 @@ local-ip-address = "0.6.1"
 chrono = "0.4.38"
 log = "0.4.20"
 
-parking_lot = { workspace = true }
+parking_lot = { workspace = true }
+once_cell = { workspace = true }

rocketmq-common/src/common.rs

@@ -33,6 +33,7 @@ pub mod consumer;
 mod faq;
 pub mod filter;
 pub mod future;
+pub mod hasher;
 pub mod macros;
 pub mod message;
 pub mod mix_all;

rocketmq-common/src/common/hasher.rs

@@ -0,0 +1,17 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+pub mod string_hasher;

rocketmq-common/src/common/hasher/string_hasher.rs

@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+use std::hash::Hasher;
+
+//Compatible with Java String's hash code
+pub struct JavaStringHasher {
+    state: i32,
+}
+
+impl Default for JavaStringHasher {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl JavaStringHasher {
+    pub fn new() -> Self {
+        JavaStringHasher { state: 0 }
+    }
+
+    pub fn hash_str(&mut self, s: &str) -> i32 {
+        if self.state == 0 && !s.is_empty() {
+            for c in s.chars() {
+                self.state = self.state.wrapping_mul(31).wrapping_add(c as i32);
+            }
+        }
+        self.state
+    }
+}
+
+impl Hasher for JavaStringHasher {
+    fn finish(&self) -> u64 {
+        self.state as u64
+    }
+
+    fn write(&mut self, bytes: &[u8]) {
+        for &byte in bytes {
+            self.state = self.state.wrapping_mul(31).wrapping_add(byte as i32);
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    #[test]
+    fn test_java_string_hasher() {
+        let mut hasher = JavaStringHasher::new();
+        let i = hasher.hash_str("hello world");
+        assert_eq!(i, 1794106052);
+    }
+}

rocketmq-common/src/common/message/message_single.rs

@@ -25,6 +25,7 @@ use bytes::{Buf, BufMut};
 
 use crate::{
     common::{
+        hasher::string_hasher::JavaStringHasher,
         message::{MessageConst, MessageTrait, MessageVersion},
         sys_flag::message_sys_flag::MessageSysFlag,
         TopicFilterType,
@@ -414,13 +415,11 @@ impl MessageExtBrokerInner {
         self.message_ext_inner.queue_offset()
     }
 
-    pub fn tags_string2tags_code(_filter: &TopicFilterType, tags: &str) -> u64 {
+    pub fn tags_string2tags_code(_filter: &TopicFilterType, tags: &str) -> i64 {
         if tags.is_empty() {
             return 0;
         }
-        let mut hasher = DefaultHasher::new();
-        tags.hash(&mut hasher);
-        hasher.finish()
+        JavaStringHasher::new().hash_str(tags) as i64
     }
 
     pub fn get_tags(&self) -> Option<String> {
@@ -444,7 +443,5 @@ pub fn tags_string2tags_code(tags: Option<&String>) -> i64 {
     if tags.is_empty() {
         return 0;
     }
-    let mut hasher = DefaultHasher::new();
-    tags.hash(&mut hasher);
-    hasher.finish() as i64
+    JavaStringHasher::new().hash_str(tags.as_str()) as i64
 }

rocketmq-common/src/common/mix_all.rs

@@ -15,6 +15,10 @@
  * limitations under the License.
  */
 
+use std::env;
+
+use once_cell::sync::Lazy;
+
 pub const ROCKETMQ_HOME_ENV: &str = "ROCKETMQ_HOME";
 pub const ROCKETMQ_HOME_PROPERTY: &str = "rocketmq.home.dir";
 pub const NAMESRV_ADDR_ENV: &str = "NAMESRV_ADDR";
@@ -67,7 +71,8 @@ pub const ZONE_MODE: &str = "__ZONE_MODE";
 pub const LOGICAL_QUEUE_MOCK_BROKER_PREFIX: &str = "__syslo__";
 pub const METADATA_SCOPE_GLOBAL: &str = "__global__";
 pub const LOGICAL_QUEUE_MOCK_BROKER_NAME_NOT_EXIST: &str = "__syslo__none__";
-pub const MULTI_PATH_SPLITTER: &str = "rocketmq.broker.multiPathSplitter";
+pub static MULTI_PATH_SPLITTER: Lazy<String> =
+    Lazy::new(|| env::var("rocketmq.broker.multiPathSplitter").unwrap_or_else(|_| ",".to_string()));
 
 pub fn is_sys_consumer_group(consumer_group: &str) -> bool {
     consumer_group.starts_with(CID_RMQ_SYS_PREFIX)

rocketmq-common/src/utils/util_all.rs

@@ -15,9 +15,16 @@
  * limitations under the License.
  */
 
-use std::path::PathBuf;
+use std::{
+    env, fs,
+    path::{Path, PathBuf},
+};
 
 use chrono::{DateTime, Datelike, TimeZone, Timelike, Utc};
+use once_cell::sync::Lazy;
+use tracing::info;
+
+use crate::common::mix_all::MULTI_PATH_SPLITTER;
 
 const HEX_ARRAY: [char; 16] = [
     '0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'A', 'B', 'C', 'D', 'E', 'F',
@@ -95,16 +102,28 @@ pub fn offset_to_file_name(offset: u64) -> String {
     format!("{:020}", offset)
 }
 
-/*pub fn ensure_dir_ok(dir: impl AsRef<PathBuf>) -> Result<(), std::io::Error> {
-    match dir.as_ref().exists() {
-        true => Ok(()),
-        false => Err(std::io::Error::new(
-            std::io::ErrorKind::NotFound,
-            format!("{:?}", dir.as_ref()),
-        )),
+pub fn ensure_dir_ok(dir_name: &str) {
+    if !dir_name.is_empty() {
+        let multi_path_splitter = MULTI_PATH_SPLITTER.as_str();
+        if dir_name.contains(multi_path_splitter) {
+            for dir in dir_name.trim().split(&multi_path_splitter) {
+                create_dir_if_not_exist(dir);
+            }
+        } else {
+            create_dir_if_not_exist(dir_name);
+        }
     }
-}*/
+}
 
+fn create_dir_if_not_exist(dir_name: &str) {
+    let path = Path::new(dir_name);
+    if !path.exists() {
+        match fs::create_dir_all(path) {
+            Ok(_) => info!("{} mkdir OK", dir_name),
+            Err(_) => info!("{} mkdir Failed", dir_name),
+        }
+    }
+}
 #[cfg(test)]
 mod tests {
     use super::*;

rocketmq-store/src/base/append_message_callback.rs

@@ -16,7 +16,7 @@
  */
 use std::{collections::HashMap, sync::Arc};
 
-use bytes::{Bytes, BytesMut};
+use bytes::{BufMut, Bytes, BytesMut};
 use rocketmq_common::{
     common::{
         attribute::cq_type::CQType,
@@ -36,7 +36,7 @@ use crate::{
     },
     config::message_store_config::MessageStoreConfig,
     log_file::{
-        commit_log::{CommitLog, CRC32_RESERVED_LEN},
+        commit_log::{CommitLog, BLANK_MAGIC_CODE, CRC32_RESERVED_LEN},
         mapped_file::MappedFile,
     },
 };
@@ -101,7 +101,7 @@ pub trait AppendMessageCallback {
 const END_FILE_MIN_BLANK_LENGTH: i32 = 4 + 4;
 
 pub(crate) struct DefaultAppendMessageCallback {
-    msg_store_item_memory: bytes::BytesMut,
+    //msg_store_item_memory: bytes::BytesMut,
     crc32_reserved_length: i32,
     message_store_config: Arc<MessageStoreConfig>,
     topic_config_table: Arc<parking_lot::Mutex<HashMap<String, TopicConfig>>>,
@@ -113,9 +113,9 @@ impl DefaultAppendMessageCallback {
         topic_config_table: Arc<parking_lot::Mutex<HashMap<String, TopicConfig>>>,
     ) -> Self {
         Self {
-            msg_store_item_memory: bytes::BytesMut::with_capacity(
+            /*            msg_store_item_memory: bytes::BytesMut::with_capacity(
                 END_FILE_MIN_BLANK_LENGTH as usize,
-            ),
+            ),*/
             crc32_reserved_length: CRC32_RESERVED_LEN,
             message_store_config,
             topic_config_table,
@@ -178,42 +178,38 @@ impl AppendMessageCallback for DefaultAppendMessageCallback {
         match MessageSysFlag::get_transaction_value(msg_inner.sys_flag()) {
             MessageSysFlag::TRANSACTION_PREPARED_TYPE
             | MessageSysFlag::TRANSACTION_ROLLBACK_TYPE => queue_offset = 0,
-            // MessageSysFlag::TRANSACTION_NOT_TYPE | MessageSysFlag::TRANSACTION_COMMIT_TYPE | _ =>
-            // {}
             _ => {}
         }
 
         if (msg_len + END_FILE_MIN_BLANK_LENGTH) > max_blank {
-            /*self.msg_store_item_memory.borrow_mut().clear();
-            self.msg_store_item_memory.borrow_mut().put_i32(max_blank);
-            self.msg_store_item_memory
-                .borrow_mut()
-                .put_i32(BLANK_MAGIC_CODE);
-            let bytes = self.msg_store_item_memory.borrow_mut().split().freeze();
-            mapped_file.append_message_bytes(&bytes);*/
+            let mut bytes = BytesMut::with_capacity(END_FILE_MIN_BLANK_LENGTH as usize);
+            bytes.put_i32(max_blank);
+            bytes.put_i32(BLANK_MAGIC_CODE);
+            mapped_file.append_message_bytes(&bytes.freeze());
             return AppendMessageResult {
                 status: AppendMessageStatus::EndOfFile,
                 wrote_offset,
                 wrote_bytes: max_blank,
                 msg_id,
                 store_timestamp: msg_inner.store_timestamp(),
                 logics_offset: queue_offset,
+                msg_num: message_num as i32,
                 ..Default::default()
             };
         }
 
         let mut pos = 4 + 4 + 4 + 4 + 4;
-        pre_encode_buffer[pos..(pos + 8)].copy_from_slice(&queue_offset.to_le_bytes());
+        pre_encode_buffer[pos..(pos + 8)].copy_from_slice(&queue_offset.to_be_bytes());
         pos += 8;
-        pre_encode_buffer[pos..(pos + 8)].copy_from_slice(&wrote_offset.to_le_bytes());
+        pre_encode_buffer[pos..(pos + 8)].copy_from_slice(&wrote_offset.to_be_bytes());
         let ip_len = if msg_inner.sys_flag() & MessageSysFlag::BORNHOST_V6_FLAG == 0 {
             4 + 4
         } else {
             16 + 4
         };
         pos += 8 + 4 + 8 + ip_len;
         pre_encode_buffer[pos..(pos + 8)]
-            .copy_from_slice(&msg_inner.store_timestamp().to_le_bytes());
+            .copy_from_slice(&msg_inner.store_timestamp().to_be_bytes());
 
         // msg_inner.encoded_buff = pre_encode_buffer;
         let bytes = Bytes::from(pre_encode_buffer);
@@ -225,6 +221,7 @@ impl AppendMessageCallback for DefaultAppendMessageCallback {
             msg_id,
             store_timestamp: msg_inner.store_timestamp(),
             logics_offset: queue_offset,
+            msg_num: message_num as i32,
             ..Default::default()
         }
     }

rocketmq-store/src/consume_queue/mapped_file_queue.rs

@@ -294,7 +294,11 @@ impl MappedFileQueue {
                     >= last_mapped_file.as_ref().unwrap().get_file_from_offset() as i64
                         + self.mapped_file_size as i64
             {
-                None
+                if return_first_on_not_found {
+                    first_mapped_file
+                } else {
+                    None
+                }
             } else {
                 let index = offset as usize / self.mapped_file_size as usize
                     - first_mapped_file.as_ref().unwrap().get_file_from_offset() as usize