From a7df129f45b509fc5152556ba93f4b1cc595eee7 Mon Sep 17 00:00:00 2001 From: "wangjiahua.wjh" Date: Thu, 27 Aug 2026 17:25:57 +0800 Subject: [PATCH] [ISSUE #10976] Reduce per-message allocation on the proxy gRPC path --- .gitignore | 5 ++ .../rocketmq/common/utils/BinaryUtil.java | 16 ++++-- .../rocketmq/common/utils/BinaryUtilTest.java | 49 +++++++++++++++++++ .../grpc/v2/producer/SendMessageActivity.java | 40 ++++++++++++--- .../v2/producer/SendMessageActivityTest.java | 18 +++++++ 5 files changed, 118 insertions(+), 10 deletions(-) create mode 100644 common/src/test/java/org/apache/rocketmq/common/utils/BinaryUtilTest.java diff --git a/.gitignore b/.gitignore index 7c29bb6beef..c63c4ac9b48 100644 --- a/.gitignore +++ b/.gitignore @@ -20,3 +20,8 @@ bazel-testlogs .vscode MODULE.bazel.lock *.flattened-pom.xml +.qoder/ +profiling-docs/ +profiling-tools/ +.gorepro/ +.gotmp/ diff --git a/common/src/main/java/org/apache/rocketmq/common/utils/BinaryUtil.java b/common/src/main/java/org/apache/rocketmq/common/utils/BinaryUtil.java index 68d15e0708a..0380633dc78 100644 --- a/common/src/main/java/org/apache/rocketmq/common/utils/BinaryUtil.java +++ b/common/src/main/java/org/apache/rocketmq/common/utils/BinaryUtil.java @@ -23,13 +23,21 @@ import org.apache.commons.codec.binary.Hex; public class BinaryUtil { - public static byte[] calculateMd5(byte[] binaryData) { - MessageDigest messageDigest = null; + /** + * MessageDigest is not thread safe, so keep one instance per thread instead of + * looking it up from the provider on every call. + */ + private static final ThreadLocal MD5_DIGEST = ThreadLocal.withInitial(() -> { try { - messageDigest = MessageDigest.getInstance("MD5"); + return MessageDigest.getInstance("MD5"); } catch (NoSuchAlgorithmException e) { - throw new RuntimeException("MD5 algorithm not found."); + throw new RuntimeException("MD5 algorithm not found.", e); } + }); + + public static byte[] calculateMd5(byte[] binaryData) { + MessageDigest messageDigest = MD5_DIGEST.get(); + messageDigest.reset(); messageDigest.update(binaryData); return messageDigest.digest(); } diff --git a/common/src/test/java/org/apache/rocketmq/common/utils/BinaryUtilTest.java b/common/src/test/java/org/apache/rocketmq/common/utils/BinaryUtilTest.java new file mode 100644 index 00000000000..5e870c3df34 --- /dev/null +++ b/common/src/test/java/org/apache/rocketmq/common/utils/BinaryUtilTest.java @@ -0,0 +1,49 @@ +/* + * 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. + */ +package org.apache.rocketmq.common.utils; + +import java.nio.charset.StandardCharsets; +import java.security.MessageDigest; +import org.apache.commons.codec.binary.Hex; +import org.junit.Test; + +import static org.junit.Assert.assertEquals; + +public class BinaryUtilTest { + + @Test + public void testGenerateMd5MatchesFreshDigestAndIsStableAcrossCalls() throws Exception { + byte[] payload = "rocketmq-md5-test".getBytes(StandardCharsets.UTF_8); + String expected = Hex.encodeHexString(MessageDigest.getInstance("MD5").digest(payload), false); + + String first = BinaryUtil.generateMd5(payload); + // interleave another digest to verify the reused per-thread instance is reset between calls + BinaryUtil.generateMd5("another-payload".getBytes(StandardCharsets.UTF_8)); + String second = BinaryUtil.generateMd5(payload); + + assertEquals(expected, first); + assertEquals(expected, second); + } + + @Test + public void testGenerateMd5FromString() throws Exception { + String body = "string-body-\u4e2d\u6587"; + String expected = Hex.encodeHexString( + MessageDigest.getInstance("MD5").digest(body.getBytes(StandardCharsets.UTF_8)), false); + assertEquals(expected, BinaryUtil.generateMd5(body)); + } +} diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/producer/SendMessageActivity.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/producer/SendMessageActivity.java index c0138cae7fa..6bdb97263fc 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/producer/SendMessageActivity.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/producer/SendMessageActivity.java @@ -29,7 +29,6 @@ import com.google.protobuf.Timestamp; import com.google.protobuf.util.Durations; import com.google.protobuf.util.Timestamps; -import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.HashSet; import java.util.List; @@ -166,7 +165,7 @@ protected void validateMessageGroup(String messageGroup) { if (maxSize <= 0) { return; } - if (messageGroup.getBytes(StandardCharsets.UTF_8).length >= maxSize) { + if (utf8Length(messageGroup) >= maxSize) { throw new GrpcProxyException(Code.ILLEGAL_MESSAGE_GROUP, "message group exceed the max size " + maxSize); } if (GrpcValidator.getInstance().containControlCharacter(messageGroup)) { @@ -214,8 +213,8 @@ protected Map buildMessageProperty(ProxyContext context, apache. if (GrpcValidator.getInstance().containControlCharacter(userPropertiesEntry.getValue())) { throw new GrpcProxyException(Code.ILLEGAL_MESSAGE_PROPERTY_KEY, "the value of property cannot contain control character"); } - userPropertySize += userPropertiesEntry.getKey().getBytes(StandardCharsets.UTF_8).length; - userPropertySize += userPropertiesEntry.getValue().getBytes(StandardCharsets.UTF_8).length; + userPropertySize += utf8Length(userPropertiesEntry.getKey()); + userPropertySize += utf8Length(userPropertiesEntry.getValue()); } MessageAccessor.setProperties(messageWithHeader, Maps.newHashMap(userProperties)); @@ -223,13 +222,13 @@ protected Map buildMessageProperty(ProxyContext context, apache. String tag = message.getSystemProperties().getTag(); GrpcValidator.getInstance().validateTag(tag); messageWithHeader.setTags(tag); - userPropertySize += tag.getBytes(StandardCharsets.UTF_8).length; + userPropertySize += utf8Length(tag); // set keys List keysList = message.getSystemProperties().getKeysList(); for (String key : keysList) { validateMessageKey(key); - userPropertySize += key.getBytes(StandardCharsets.UTF_8).length; + userPropertySize += utf8Length(key); } if (keysList.size() > 0) { messageWithHeader.setKeys(keysList); @@ -311,6 +310,35 @@ protected Map buildMessageProperty(ProxyContext context, apache. return messageWithHeader.getProperties(); } + /** + * Length of the UTF-8 encoding of {@code str} without materializing the byte array. + * Matches {@code str.getBytes(StandardCharsets.UTF_8).length}, including the + * single-byte replacement for unpaired surrogates. + */ + static int utf8Length(String str) { + int len = 0; + for (int i = 0; i < str.length(); i++) { + char c = str.charAt(i); + if (c < 0x80) { + len += 1; + } else if (c < 0x800) { + len += 2; + } else if (Character.isHighSurrogate(c)) { + if (i + 1 < str.length() && Character.isLowSurrogate(str.charAt(i + 1))) { + len += 4; + i++; + } else { + len += 1; + } + } else if (Character.isLowSurrogate(c)) { + len += 1; + } else { + len += 3; + } + } + return len; + } + protected void fillDelayMessageProperty(apache.rocketmq.v2.Message message, org.apache.rocketmq.common.message.Message messageWithHeader) { if (message.getSystemProperties().hasDeliveryTimestamp()) { Timestamp deliveryTimestamp = message.getSystemProperties().getDeliveryTimestamp(); diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/producer/SendMessageActivityTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/producer/SendMessageActivityTest.java index f9761e299af..2501c34813d 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/producer/SendMessageActivityTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/producer/SendMessageActivityTest.java @@ -28,6 +28,7 @@ import com.google.protobuf.ByteString; import com.google.protobuf.util.Durations; import com.google.protobuf.util.Timestamps; +import java.nio.charset.StandardCharsets; import java.time.Duration; import java.util.HashMap; import java.util.Map; @@ -89,6 +90,23 @@ public void before() throws Throwable { this.sendMessageActivity = new SendMessageActivity(messagingProcessor, grpcClientSettingsManager, grpcChannelManager); } + @Test + public void testUtf8Length() { + String[] samples = { + "", + "abc123", + "\u4e2d\u6587\u5c5e\u6027\u503c", + "mixed\u4e2d\u6587and😀emoji", + "\uD83D\uDE00", + "\uD800", + "abc\uDC00def", + "߿ࠀ" + }; + for (String sample : samples) { + assertEquals(sample, sample.getBytes(StandardCharsets.UTF_8).length, SendMessageActivity.utf8Length(sample)); + } + } + @Test public void sendMessage() throws Exception { String msgId = MessageClientIDSetter.createUniqID();