opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-10-29 17:25:35 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #4203] Support zstd/lz4 compression to improve send/receive performance

Browse Source
This commit is contained in:
wangfan
2022-04-28 10:19:52 +08:00
parent a492592d77
commit 5584a78977
25 changed files with 709 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/src/main/java/org/apache/rocketmq/client/impl/ClientRemotingProcessor.java
+6 -2
View File
@@ -28,6 +28,8 @@ import org.apache.rocketmq.client.log.ClientLogger;
import org.apache.rocketmq.client.producer.RequestFutureHolder;
import org.apache.rocketmq.client.producer.RequestResponseFuture;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.compression.Compressor;
import org.apache.rocketmq.common.compression.CompressorFactory;
import org.apache.rocketmq.common.message.MessageAccessor;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.common.message.MessageDecoder;
@@ -245,9 +247,11 @@ public class ClientRemotingProcessor extends AsyncNettyRequestProcessor implemen
}
byte[] body = request.getBody();
if ((requestHeader.getSysFlag() & MessageSysFlag.COMPRESSED_FLAG) == MessageSysFlag.COMPRESSED_FLAG) {
int sysFlag = requestHeader.getSysFlag();
if ((sysFlag & MessageSysFlag.COMPRESSED_FLAG) == MessageSysFlag.COMPRESSED_FLAG) {
try {
body = UtilAll.uncompress(body);
Compressor compressor = CompressorFactory.getCompressor(MessageSysFlag.getCompressionType(sysFlag));
body = compressor.decompress(body);
} catch (IOException e) {
log.warn("err when uncompress constant", e);
}
client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java
+32 -12
View File
@@ -67,7 +67,9 @@ import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.ServiceState;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.compression.CompressionType;
import org.apache.rocketmq.common.compression.Compressor;
import org.apache.rocketmq.common.compression.CompressorFactory;
import org.apache.rocketmq.common.help.FAQUrl;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageAccessor;
@@ -109,10 +111,14 @@ public class DefaultMQProducerImpl implements MQProducerInner {
private ServiceState serviceState = ServiceState.CREATE_JUST;
private MQClientInstance mQClientFactory;
private ArrayList<CheckForbiddenHook> checkForbiddenHookList = new ArrayList<CheckForbiddenHook>();
private int zipCompressLevel = Integer.parseInt(System.getProperty(MixAll.MESSAGE_COMPRESS_LEVEL, "5"));
private MQFaultStrategy mqFaultStrategy = new MQFaultStrategy();
private ExecutorService asyncSenderExecutor;
// compression related
private int compressLevel = Integer.parseInt(System.getProperty(MixAll.MESSAGE_COMPRESS_LEVEL, "5"));
private CompressionType compressType = CompressionType.of(System.getProperty(MixAll.MESSAGE_COMPRESS_TYPE, "ZLIB"));
private final Compressor compressor = CompressorFactory.getCompressor(compressType);
public DefaultMQProducerImpl(final DefaultMQProducer defaultMQProducer) {
this(defaultMQProducer, null);
}
@@ -530,7 +536,7 @@ public class DefaultMQProducerImpl implements MQProducerInner {
}
private void validateNameServerSetting() throws MQClientException {
List<String> nsList = this.getmQClientFactory().getMQClientAPIImpl().getNameServerAddressList();
List<String> nsList = this.getMqClientFactory().getMQClientAPIImpl().getNameServerAddressList();
if (null == nsList || nsList.isEmpty()) {
throw new MQClientException(
"No name server address, please set it." + FAQUrl.suggestTodo(FAQUrl.NAME_SERVER_ADDR_NOT_EXIST_URL), null).setResponseCode(ClientErrorCode.NO_NAME_SERVER_EXCEPTION);
@@ -724,6 +730,7 @@ public class DefaultMQProducerImpl implements MQProducerInner {
boolean msgBodyCompressed = false;
if (this.tryToCompressMessage(msg)) {
sysFlag |= MessageSysFlag.COMPRESSED_FLAG;
sysFlag |= compressType.getCompressionFlag();
msgBodyCompressed = true;
}
@@ -886,20 +893,25 @@ public class DefaultMQProducerImpl implements MQProducerInner {
throw new MQClientException("The broker[" + mq.getBrokerName() + "] not exist", null);
}
public MQClientInstance getMqClientFactory() {
return mQClientFactory;
}
@Deprecated
public MQClientInstance getmQClientFactory() {
return mQClientFactory;
}
private boolean tryToCompressMessage(final Message msg) {
if (msg instanceof MessageBatch) {
//batch dose not support compressing right now
//batch does not support compressing right now
return false;
}
byte[] body = msg.getBody();
if (body != null) {
if (body.length >= this.defaultMQProducer.getCompressMsgBodyOverHowmuch()) {
try {
byte[] data = UtilAll.compress(body, zipCompressLevel);
byte[] data = compressor.compress(body, compressLevel);
if (data != null) {
msg.setBody(data);
return true;
@@ -1561,16 +1573,16 @@ public class DefaultMQProducerImpl implements MQProducerInner {
private void prepareSendRequest(final Message msg, long timeout) {
String correlationId = CorrelationIdUtil.createCorrelationId();
String requestClientId = this.getmQClientFactory().getClientId();
String requestClientId = this.getMqClientFactory().getClientId();
MessageAccessor.putProperty(msg, MessageConst.PROPERTY_CORRELATION_ID, correlationId);
MessageAccessor.putProperty(msg, MessageConst.PROPERTY_MESSAGE_REPLY_TO_CLIENT, requestClientId);
MessageAccessor.putProperty(msg, MessageConst.PROPERTY_MESSAGE_TTL, String.valueOf(timeout));
boolean hasRouteData = this.getmQClientFactory().getTopicRouteTable().containsKey(msg.getTopic());
boolean hasRouteData = this.getMqClientFactory().getTopicRouteTable().containsKey(msg.getTopic());
if (!hasRouteData) {
long beginTimestamp = System.currentTimeMillis();
this.tryToFindTopicPublishInfo(msg.getTopic());
this.getmQClientFactory().sendHeartbeatToAllBrokerWithLock();
this.getMqClientFactory().sendHeartbeatToAllBrokerWithLock();
long cost = System.currentTimeMillis() - beginTimestamp;
if (cost > 500) {
log.warn("prepare send request for <{}> cost {} ms", msg.getTopic(), cost);
@@ -1582,12 +1594,20 @@ public class DefaultMQProducerImpl implements MQProducerInner {
return topicPublishInfoTable;
}
public int getZipCompressLevel() {
return zipCompressLevel;
public int getCompressLevel() {
return compressLevel;
}
public void setZipCompressLevel(int zipCompressLevel) {
this.zipCompressLevel = zipCompressLevel;
public void setCompressLevel(int compressLevel) {
this.compressLevel = compressLevel;
}
public CompressionType getCompressType() {
return compressType;
}
public void setCompressType(CompressionType compressType) {
this.compressType = compressType;
}
public ServiceState getServiceState() {
client/src/main/java/org/apache/rocketmq/client/trace/AsyncTraceDispatcher.java
+1 -1
View File
@@ -438,7 +438,7 @@ public class AsyncTraceDispatcher implements TraceDispatcher {
TopicPublishInfo topicPublishInfo = producer.getTopicPublishInfoTable().get(topic);
if (null == topicPublishInfo || !topicPublishInfo.ok()) {
producer.getTopicPublishInfoTable().putIfAbsent(topic, new TopicPublishInfo());
producer.getmQClientFactory().updateTopicRouteInfoFromNameServer(topic);
producer.getMqClientFactory().updateTopicRouteInfoFromNameServer(topic);
topicPublishInfo = producer.getTopicPublishInfoTable().get(topic);
}
if (topicPublishInfo.isHaveTopicRouterInfo() || topicPublishInfo.ok()) {
client/src/main/java/org/apache/rocketmq/client/trace/hook/EndTransactionTraceHookImpl.java
+1 -1
View File
@@ -63,7 +63,7 @@ public class EndTransactionTraceHookImpl implements EndTransactionHook {
traceBean.setKeys(context.getMessage().getKeys());
traceBean.setStoreHost(context.getBrokerAddr());
traceBean.setMsgType(MessageType.Trans_msg_Commit);
traceBean.setClientHost(((AsyncTraceDispatcher)localDispatcher).getHostProducer().getmQClientFactory().getClientId());
traceBean.setClientHost(((AsyncTraceDispatcher)localDispatcher).getHostProducer().getMqClientFactory().getClientId());
traceBean.setMsgId(context.getMsgId());
traceBean.setTransactionState(context.getTransactionState());
traceBean.setTransactionId(context.getTransactionId());
client/src/test/java/org/apache/rocketmq/client/producer/DefaultMQProducerTest.java
+2 -2
View File
@@ -102,7 +102,7 @@ public class DefaultMQProducerTest {
field.setAccessible(true);
field.set(mQClientFactory, mQClientAPIImpl);
producer.getDefaultMQProducerImpl().getmQClientFactory().registerProducer(producerGroupTemp, producer.getDefaultMQProducerImpl());
producer.getDefaultMQProducerImpl().getMqClientFactory().registerProducer(producerGroupTemp, producer.getDefaultMQProducerImpl());
when(mQClientAPIImpl.sendMessage(anyString(), anyString(), any(Message.class), any(SendMessageRequestHeader.class), anyLong(), any(CommunicationMode.class),
nullable(SendMessageContext.class), any(DefaultMQProducerImpl.class))).thenCallRealMethod();
@@ -358,7 +358,7 @@ public class DefaultMQProducerTest {
producer.setCallbackExecutor(customized);
NettyRemotingClient remotingClient = (NettyRemotingClient) producer.getDefaultMQProducerImpl()
.getmQClientFactory().getMQClientAPIImpl().getRemotingClient();
.getMqClientFactory().getMQClientAPIImpl().getRemotingClient();
assertThat(remotingClient.getCallbackExecutor()).isEqualTo(customized);
}
client/src/test/java/org/apache/rocketmq/client/trace/DefaultMQConsumerWithTraceTest.java
+1 -1
View File
@@ -217,7 +217,7 @@ public class DefaultMQConsumerWithTraceTest {
@Test
public void testPullMessage_WithTrace_Success() throws InterruptedException, RemotingException, MQBrokerException, MQClientException {
traceProducer.getDefaultMQProducerImpl().getmQClientFactory().registerProducer(producerGroupTraceTemp, traceProducer.getDefaultMQProducerImpl());
traceProducer.getDefaultMQProducerImpl().getMqClientFactory().registerProducer(producerGroupTraceTemp, traceProducer.getDefaultMQProducerImpl());
final CountDownLatch countDownLatch = new CountDownLatch(1);
final AtomicReference<MessageExt> messageAtomic = new AtomicReference<MessageExt>();
client/src/test/java/org/apache/rocketmq/client/trace/DefaultMQLitePullConsumerWithTraceTest.java
+1 -1
View File
@@ -218,7 +218,7 @@ public class DefaultMQLitePullConsumerWithTraceTest {
field.setAccessible(true);
field.set(litePullConsumerImpl, offsetStore);
traceProducer.getDefaultMQProducerImpl().getmQClientFactory().registerProducer(producerGroupTraceTemp, traceProducer.getDefaultMQProducerImpl());
traceProducer.getDefaultMQProducerImpl().getMqClientFactory().registerProducer(producerGroupTraceTemp, traceProducer.getDefaultMQProducerImpl());
when(mQClientFactory.getMQClientAPIImpl().pullMessage(anyString(), any(PullMessageRequestHeader.class),
anyLong(), any(CommunicationMode.class), nullable(PullCallback.class)))
client/src/test/java/org/apache/rocketmq/client/trace/DefaultMQProducerWithOpenTracingTest.java
+2 -2
View File
@@ -100,7 +100,7 @@ public class DefaultMQProducerWithOpenTracingTest {
field.setAccessible(true);
field.set(mQClientFactory, mQClientAPIImpl);
producer.getDefaultMQProducerImpl().getmQClientFactory().registerProducer(producerGroupTemp, producer.getDefaultMQProducerImpl());
producer.getDefaultMQProducerImpl().getMqClientFactory().registerProducer(producerGroupTemp, producer.getDefaultMQProducerImpl());
when(mQClientAPIImpl.sendMessage(anyString(), anyString(), any(Message.class), any(SendMessageRequestHeader.class), anyLong(), any(CommunicationMode.class),
nullable(SendMessageContext.class), any(DefaultMQProducerImpl.class))).thenCallRealMethod();
@@ -112,7 +112,7 @@ public class DefaultMQProducerWithOpenTracingTest {
@Test
public void testSendMessageSync_WithTrace_Success() throws RemotingException, InterruptedException, MQBrokerException, MQClientException {
producer.getDefaultMQProducerImpl().getmQClientFactory().registerProducer(producerGroupTraceTemp, producer.getDefaultMQProducerImpl());
producer.getDefaultMQProducerImpl().getMqClientFactory().registerProducer(producerGroupTraceTemp, producer.getDefaultMQProducerImpl());
when(mQClientAPIImpl.getTopicRouteInfoFromNameServer(anyString(), anyLong())).thenReturn(createTopicRoute());
producer.send(message);
assertThat(tracer.finishedSpans().size()).isEqualTo(1);
client/src/test/java/org/apache/rocketmq/client/trace/DefaultMQProducerWithTraceTest.java
+2 -2
View File
@@ -109,7 +109,7 @@ public class DefaultMQProducerWithTraceTest {
field.setAccessible(true);
field.set(mQClientFactory, mQClientAPIImpl);
producer.getDefaultMQProducerImpl().getmQClientFactory().registerProducer(producerGroupTemp, producer.getDefaultMQProducerImpl());
producer.getDefaultMQProducerImpl().getMqClientFactory().registerProducer(producerGroupTemp, producer.getDefaultMQProducerImpl());
when(mQClientAPIImpl.sendMessage(anyString(), anyString(), any(Message.class), any(SendMessageRequestHeader.class), anyLong(), any(CommunicationMode.class),
nullable(SendMessageContext.class), any(DefaultMQProducerImpl.class))).thenCallRealMethod();
@@ -121,7 +121,7 @@ public class DefaultMQProducerWithTraceTest {
@Test
public void testSendMessageSync_WithTrace_Success() throws RemotingException, InterruptedException, MQBrokerException, MQClientException {
traceProducer.getDefaultMQProducerImpl().getmQClientFactory().registerProducer(producerGroupTraceTemp, traceProducer.getDefaultMQProducerImpl());
traceProducer.getDefaultMQProducerImpl().getMqClientFactory().registerProducer(producerGroupTraceTemp, traceProducer.getDefaultMQProducerImpl());
when(mQClientAPIImpl.getTopicRouteInfoFromNameServer(anyString(), anyLong())).thenReturn(createTopicRoute());
final CountDownLatch countDownLatch = new CountDownLatch(1);
try {
client/src/test/java/org/apache/rocketmq/client/trace/TransactionMQProducerWithOpenTracingTest.java
+2 -2
View File
@@ -118,7 +118,7 @@ public class TransactionMQProducerWithOpenTracingTest {
field.setAccessible(true);
field.set(mQClientFactory, mQClientAPIImpl);
producer.getDefaultMQProducerImpl().getmQClientFactory().registerProducer(producerGroupTemp, producer.getDefaultMQProducerImpl());
producer.getDefaultMQProducerImpl().getMqClientFactory().registerProducer(producerGroupTemp, producer.getDefaultMQProducerImpl());
when(mQClientAPIImpl.sendMessage(anyString(), anyString(), any(Message.class), any(SendMessageRequestHeader.class), anyLong(), any(CommunicationMode.class),
nullable(SendMessageContext.class), any(DefaultMQProducerImpl.class))).thenCallRealMethod();
@@ -130,7 +130,7 @@ public class TransactionMQProducerWithOpenTracingTest {
@Test
public void testSendMessageSync_WithTrace_Success() throws RemotingException, InterruptedException, MQBrokerException, MQClientException {
producer.getDefaultMQProducerImpl().getmQClientFactory().registerProducer(producerGroupTraceTemp, producer.getDefaultMQProducerImpl());
producer.getDefaultMQProducerImpl().getMqClientFactory().registerProducer(producerGroupTraceTemp, producer.getDefaultMQProducerImpl());
when(mQClientAPIImpl.getTopicRouteInfoFromNameServer(anyString(), anyLong())).thenReturn(createTopicRoute());
producer.sendMessageInTransaction(message, null);
client/src/test/java/org/apache/rocketmq/client/trace/TransactionMQProducerWithTraceTest.java
+2 -2
View File
@@ -128,7 +128,7 @@ public class TransactionMQProducerWithTraceTest {
field.setAccessible(true);
field.set(mQClientFactory, mQClientAPIImpl);
producer.getDefaultMQProducerImpl().getmQClientFactory().registerProducer(producerGroupTemp, producer.getDefaultMQProducerImpl());
producer.getDefaultMQProducerImpl().getMqClientFactory().registerProducer(producerGroupTemp, producer.getDefaultMQProducerImpl());
Field fieldHooks = DefaultMQProducerImpl.class.getDeclaredField("endTransactionHookList");
fieldHooks.setAccessible(true);
@@ -146,7 +146,7 @@ public class TransactionMQProducerWithTraceTest {
@Test
public void testSendMessageSync_WithTrace_Success() throws RemotingException, InterruptedException, MQBrokerException, MQClientException {
traceProducer.getDefaultMQProducerImpl().getmQClientFactory().registerProducer(producerGroupTraceTemp, traceProducer.getDefaultMQProducerImpl());
traceProducer.getDefaultMQProducerImpl().getMqClientFactory().registerProducer(producerGroupTraceTemp, traceProducer.getDefaultMQProducerImpl());
when(mQClientAPIImpl.getTopicRouteInfoFromNameServer(anyString(), anyLong())).thenReturn(createTopicRoute());
final AtomicReference<EndTransactionContext> context = new AtomicReference<EndTransactionContext>();
doAnswer(new Answer() {
common/pom.xml
+8
View File
@@ -40,5 +40,13 @@
<groupId>commons-validator</groupId>
<artifactId>commons-validator</artifactId>
</dependency>
<dependency>
<groupId>com.github.luben</groupId>
<artifactId>zstd-jni</artifactId>
</dependency>
<dependency>
<groupId>org.lz4</groupId>
<artifactId>lz4-java</artifactId>
</dependency>
</dependencies>
</project>
common/src/main/java/org/apache/rocketmq/common/MixAll.java
+1
View File
@@ -50,6 +50,7 @@ public class MixAll {
public static final String ROCKETMQ_HOME_PROPERTY = "rocketmq.home.dir";
public static final String NAMESRV_ADDR_ENV = "NAMESRV_ADDR";
public static final String NAMESRV_ADDR_PROPERTY = "rocketmq.namesrv.addr";
public static final String MESSAGE_COMPRESS_TYPE = "rocketmq.message.compressType";
public static final String MESSAGE_COMPRESS_LEVEL = "rocketmq.message.compressLevel";
public static final String DEFAULT_NAMESRV_ADDR_LOOKUP = "jmenv.tbsite.net";
public static final String WS_DOMAIN_NAME = System.getProperty("rocketmq.namesrv.domain", DEFAULT_NAMESRV_ADDR_LOOKUP);
common/src/main/java/org/apache/rocketmq/common/UtilAll.java
+8
View File
@@ -295,6 +295,10 @@ public class UtilAll {
return (byte) "0123456789ABCDEF".indexOf(c);
}
/**
* use {@link org.apache.rocketmq.common.compression.Compressor#decompress(byte[])} instead.
*/
@Deprecated
public static byte[] uncompress(final byte[] src) throws IOException {
byte[] result = src;
byte[] uncompressData = new byte[src.length];
@@ -335,6 +339,10 @@ public class UtilAll {
return result;
}
/**
* use {@link org.apache.rocketmq.common.compression.Compressor#compress(byte[], int)} instead.
*/
@Deprecated
public static byte[] compress(final byte[] src, final int level) throws IOException {
byte[] result = src;
ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream(src.length);
common/src/main/java/org/apache/rocketmq/common/compression/CompressionType.java
+90
View File
@@ -0,0 +1,90 @@
/*
* 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.compression;
import org.apache.rocketmq.common.sysflag.MessageSysFlag;
public enum CompressionType {
/**
* Compression types number can be extended to seven {@link MessageSysFlag}
*
* Benchmarks from https://github.com/facebook/zstd
*
* | Compressor | Ratio | Compression | Decompress |
* |----------------|---------|-------------|------------|
* | zstd 1.5.1 | 2.887 | 530 MB/s | 1700 MB/s |
* | zlib 1.2.11 | 2.743 | 95 MB/s | 400 MB/s |
* | lz4 1.9.3 | 2.101 | 740 MB/s | 4500 MB/s |
*
*/
LZ4(1),
ZSTD(2),
ZLIB(3);
private final int value;
CompressionType(int value) {
this.value = value;
}
public int getValue() {
return value;
}
public static CompressionType of(String name) {
switch (name.trim().toUpperCase()) {
case "LZ4":
return CompressionType.LZ4;
case "ZSTD":
return CompressionType.ZSTD;
case "ZLIB":
return CompressionType.ZLIB;
default:
throw new RuntimeException("Unsupported compress type name: " + name);
}
}
public static CompressionType findByValue(int value) {
switch (value) {
case 1:
return LZ4;
case 2:
return ZSTD;
case 0: // To be compatible for older versions without compression type
case 3:
return ZLIB;
default:
throw new RuntimeException("Unknown compress type value: " + value);
}
}
public int getCompressionFlag() {
switch (value) {
case 1:
return MessageSysFlag.COMPRESSION_LZ4_TYPE;
case 2:
return MessageSysFlag.COMPRESSION_ZSTD_TYPE;
case 3:
return MessageSysFlag.COMPRESSION_ZLIB_TYPE;
default:
throw new RuntimeException("Unsupported compress type flag: " + value);
}
}
}
common/src/main/java/org/apache/rocketmq/common/compression/Compressor.java
+42
View File
@@ -0,0 +1,42 @@
/*
* 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.compression;
import java.io.IOException;
public interface Compressor {
/**
* Compress message by different compressor.
*
* @param src bytes ready to compress
* @param level compression level used to balance compression rate and time consumption
* @return compressed byte data
* @throws IOException
*/
byte[] compress(byte[] src, int level) throws IOException;
/**
* Decompress message by different compressor.
*
* @param src bytes ready to decompress
* @return decompressed byte data
* @throws IOException
*/
byte[] decompress(byte[] src) throws IOException;
}
common/src/main/java/org/apache/rocketmq/common/compression/CompressorFactory.java
+36
View File
@@ -0,0 +1,36 @@
/*
* 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.compression;
import java.util.EnumMap;
public class CompressorFactory {
private static final EnumMap<CompressionType, Compressor> COMPRESSORS;
static {
COMPRESSORS = new EnumMap<>(CompressionType.class);
COMPRESSORS.put(CompressionType.LZ4, new Lz4Compressor());
COMPRESSORS.put(CompressionType.ZSTD, new ZstdCompressor());
COMPRESSORS.put(CompressionType.ZLIB, new ZlibCompressor());
}
public static Compressor getCompressor(CompressionType type) {
return COMPRESSORS.get(type);
}
}
common/src/main/java/org/apache/rocketmq/common/compression/Lz4Compressor.java
+86
View File
@@ -0,0 +1,86 @@
/*
* 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.compression;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import net.jpountz.lz4.LZ4FrameInputStream;
import net.jpountz.lz4.LZ4FrameOutputStream;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
public class Lz4Compressor implements Compressor {
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.COMMON_LOGGER_NAME);
@Override
public byte[] compress(byte[] src, int level) throws IOException {
byte[] result = src;
ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream(src.length);
LZ4FrameOutputStream outputStream = new LZ4FrameOutputStream(byteArrayOutputStream);
try {
outputStream.write(src);
outputStream.flush();
outputStream.close();
result = byteArrayOutputStream.toByteArray();
} catch (IOException e) {
log.error("Failed to compress data by lz4", e);
throw e;
} finally {
try {
byteArrayOutputStream.close();
} catch (IOException ignored) {
}
}
return result;
}
@Override
public byte[] decompress(byte[] src) throws IOException {
byte[] result = src;
byte[] uncompressData = new byte[src.length];
ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(src);
LZ4FrameInputStream lz4InputStream = new LZ4FrameInputStream(byteArrayInputStream);
ByteArrayOutputStream resultOutputStream = new ByteArrayOutputStream(src.length);
try {
while (true) {
int len = lz4InputStream.read(uncompressData, 0, uncompressData.length);
if (len <= 0) {
break;
}
resultOutputStream.write(uncompressData, 0, len);
}
resultOutputStream.flush();
resultOutputStream.close();
result = resultOutputStream.toByteArray();
} catch (IOException e) {
throw e;
} finally {
try {
lz4InputStream.close();
byteArrayInputStream.close();
} catch (IOException e) {
log.warn("Failed to close the lz4 compress stream ", e);
}
}
return result;
}
}
common/src/main/java/org/apache/rocketmq/common/compression/ZlibCompressor.java
+98
View File
@@ -0,0 +1,98 @@
/*
* 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.compression;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.zip.DeflaterOutputStream;
import java.util.zip.InflaterInputStream;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
public class ZlibCompressor implements Compressor {
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.COMMON_LOGGER_NAME);
@Override
public byte[] compress(byte[] src, int level) throws IOException {
byte[] result = src;
ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream(src.length);
java.util.zip.Deflater defeater = new java.util.zip.Deflater(level);
DeflaterOutputStream deflaterOutputStream = new DeflaterOutputStream(byteArrayOutputStream, defeater);
try {
deflaterOutputStream.write(src);
deflaterOutputStream.finish();
deflaterOutputStream.close();
result = byteArrayOutputStream.toByteArray();
} catch (IOException e) {
defeater.end();
throw e;
} finally {
try {
byteArrayOutputStream.close();
} catch (IOException ignored) {
}
defeater.end();
}
return result;
}
@Override
public byte[] decompress(byte[] src) throws IOException {
byte[] result = src;
byte[] uncompressData = new byte[src.length];
ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(src);
InflaterInputStream inflaterInputStream = new InflaterInputStream(byteArrayInputStream);
ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream(src.length);
try {
while (true) {
int len = inflaterInputStream.read(uncompressData, 0, uncompressData.length);
if (len <= 0) {
break;
}
byteArrayOutputStream.write(uncompressData, 0, len);
}
byteArrayOutputStream.flush();
result = byteArrayOutputStream.toByteArray();
} catch (IOException e) {
throw e;
} finally {
try {
byteArrayInputStream.close();
} catch (IOException e) {
log.error("Failed to close the stream", e);
}
try {
inflaterInputStream.close();
} catch (IOException e) {
log.error("Failed to close the stream", e);
}
try {
byteArrayOutputStream.close();
} catch (IOException e) {
log.error("Failed to close the stream", e);
}
}
return result;
}
}
common/src/main/java/org/apache/rocketmq/common/compression/ZstdCompressor.java
+86
View File
@@ -0,0 +1,86 @@
/*
* 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.compression;
import com.github.luben.zstd.ZstdInputStream;
import com.github.luben.zstd.ZstdOutputStream;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
public class ZstdCompressor implements Compressor {
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.COMMON_LOGGER_NAME);
@Override
public byte[] compress(byte[] src, int level) throws IOException {
byte[] result = src;
ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream(src.length);
ZstdOutputStream outputStream = new ZstdOutputStream(byteArrayOutputStream, level);
try {
outputStream.write(src);
outputStream.flush();
outputStream.close();
result = byteArrayOutputStream.toByteArray();
} catch (IOException e) {
log.error("Failed to compress data by zstd", e);
throw e;
} finally {
try {
byteArrayOutputStream.close();
} catch (IOException ignored) {
}
}
return result;
}
@Override
public byte[] decompress(byte[] src) throws IOException {
byte[] result = src;
byte[] uncompressData = new byte[src.length];
ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(src);
ZstdInputStream zstdInputStream = new ZstdInputStream(byteArrayInputStream);
ByteArrayOutputStream resultOutputStream = new ByteArrayOutputStream(src.length);
try {
while (true) {
int len = zstdInputStream.read(uncompressData, 0, uncompressData.length);
if (len <= 0) {
break;
}
resultOutputStream.write(uncompressData, 0, len);
}
resultOutputStream.flush();
resultOutputStream.close();
result = resultOutputStream.toByteArray();
} catch (IOException e) {
throw e;
} finally {
try {
zstdInputStream.close();
byteArrayInputStream.close();
} catch (IOException e) {
log.warn("Failed to close the zstd compress stream", e);
}
}
return result;
}
}
common/src/main/java/org/apache/rocketmq/common/message/MessageDecoder.java
+6 -2
View File
@@ -29,6 +29,8 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.compression.Compressor;
import org.apache.rocketmq.common.compression.CompressorFactory;
import org.apache.rocketmq.common.sysflag.MessageSysFlag;
public class MessageDecoder {
@@ -163,7 +165,8 @@ public class MessageDecoder {
int storehostAddressLength = (sysFlag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0 ? 8 : 20;
byte[] newBody = messageExt.getBody();
if (needCompress && (sysFlag & MessageSysFlag.COMPRESSED_FLAG) == MessageSysFlag.COMPRESSED_FLAG) {
newBody = UtilAll.compress(body, 5);
Compressor compressor = CompressorFactory.getCompressor(MessageSysFlag.getCompressionType(sysFlag));
newBody = compressor.compress(body, 5);
}
int bodyLength = newBody.length;
int storeSize = messageExt.getStoreSize();
@@ -347,7 +350,8 @@ public class MessageDecoder {
// uncompress body
if (deCompressBody && (sysFlag & MessageSysFlag.COMPRESSED_FLAG) == MessageSysFlag.COMPRESSED_FLAG) {
body = UtilAll.uncompress(body);
Compressor compressor = CompressorFactory.getCompressor(MessageSysFlag.getCompressionType(sysFlag));
body = compressor.decompress(body);
}
msgExt.setBody(body);
common/src/main/java/org/apache/rocketmq/common/sysflag/MessageSysFlag.java
+25
View File
@@ -16,7 +16,21 @@
*/
package org.apache.rocketmq.common.sysflag;
import org.apache.rocketmq.common.compression.CompressionType;
public class MessageSysFlag {
/**
* Meaning of each bit in the system flag
*
* | bit | 7 | 6 | 5 | 4 | 3 | 2 | 1 | 0 |
* |--------|---|---|-----------|----------|-------------|------------------|------------------|------------------|
* | byte 1 | | | STOREHOST | BORNHOST | TRANSACTION | TRANSACTION | MULTI_TAGS | COMPRESSED |
* | byte 2 | | | | | | COMPRESSION_TYPE | COMPRESSION_TYPE | COMPRESSION_TYPE |
* | byte 3 | | | | | | | | |
* | byte 4 | | | | | | | | |
*
*/
public final static int COMPRESSED_FLAG = 0x1;
public final static int MULTI_TAGS_FLAG = 0x1 << 1;
public final static int TRANSACTION_NOT_TYPE = 0;
@@ -26,6 +40,12 @@ public class MessageSysFlag {
public final static int BORNHOST_V6_FLAG = 0x1 << 4;
public final static int STOREHOSTADDRESS_V6_FLAG = 0x1 << 5;
// COMPRESSION_TYPE
public final static int COMPRESSION_LZ4_TYPE = 0x1 << 8;
public final static int COMPRESSION_ZSTD_TYPE = 0x2 << 8;
public final static int COMPRESSION_ZLIB_TYPE = 0x3 << 8;
public final static int COMPRESSION_TYPE_COMPARATOR = 0x7 << 8;
public static int getTransactionValue(final int flag) {
return flag & TRANSACTION_ROLLBACK_TYPE;
}
@@ -38,4 +58,9 @@ public class MessageSysFlag {
return flag & (~COMPRESSED_FLAG);
}
// To match the compression type
public static CompressionType getCompressionType(final int flag) {
return CompressionType.findByValue((flag & COMPRESSION_TYPE_COMPARATOR) >> 8);
}
}
common/src/test/java/org/apache/rocketmq/common/compression/CompressionTest.java
+106
View File
@@ -0,0 +1,106 @@
/*
* 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.compression;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Random;
import org.apache.commons.lang3.RandomStringUtils;
import org.junit.Before;
import org.junit.Test;
import static org.assertj.core.api.Assertions.assertThat;
public class CompressionTest {
private int level;
private Compressor zstd;
private Compressor zlib;
private Compressor lz4;
@Before
public void setUp() {
level = 5;
zstd = CompressorFactory.getCompressor(CompressionType.ZSTD);
zlib = CompressorFactory.getCompressor(CompressionType.ZLIB);
lz4 = CompressorFactory.getCompressor(CompressionType.LZ4);
}
@Test
public void testCompressionZlib() throws IOException {
assertThat(CompressionType.of("zlib")).isEqualTo(CompressionType.ZLIB);
assertThat(CompressionType.of(" ZLiB ")).isEqualTo(CompressionType.ZLIB);
assertThat(CompressionType.of("ZLIB")).isEqualTo(CompressionType.ZLIB);
int randomKB = 4096;
Random random = new Random();
for (int i = 0; i < 5; ++i) {
String message = RandomStringUtils.randomAlphanumeric(random.nextInt(randomKB) * 1024);
byte[] srcBytes = message.getBytes(StandardCharsets.UTF_8);
byte[] compressed = zlib.compress(srcBytes, level);
byte[] decompressed = zlib.decompress(compressed);
// compression ratio may be negative for some random string data
// assertThat(compressed.length).isLessThan(srcBytes.length);
assertThat(decompressed).isEqualTo(srcBytes);
assertThat(new String(decompressed)).isEqualTo(message);
}
}
@Test
public void testCompressionZstd() throws IOException {
assertThat(CompressionType.of("zstd")).isEqualTo(CompressionType.ZSTD);
assertThat(CompressionType.of("ZStd ")).isEqualTo(CompressionType.ZSTD);
assertThat(CompressionType.of("ZSTD")).isEqualTo(CompressionType.ZSTD);
int randomKB = 4096;
Random random = new Random();
for (int i = 0; i < 5; ++i) {
String message = RandomStringUtils.randomAlphanumeric(random.nextInt(randomKB) * 1024);
byte[] srcBytes = message.getBytes(StandardCharsets.UTF_8);
byte[] compressed = zstd.compress(srcBytes, level);
byte[] decompressed = zstd.decompress(compressed);
// compression ratio may be negative for some random string data
// assertThat(compressed.length).isLessThan(srcBytes.length);
assertThat(decompressed).isEqualTo(srcBytes);
assertThat(new String(decompressed)).isEqualTo(message);
}
}
@Test
public void testCompressionLz4() throws IOException {
assertThat(CompressionType.of("lz4")).isEqualTo(CompressionType.LZ4);
int randomKB = 4096;
Random random = new Random();
for (int i = 0; i < 5; ++i) {
String message = RandomStringUtils.randomAlphanumeric(random.nextInt(randomKB) * 1024);
byte[] srcBytes = message.getBytes(StandardCharsets.UTF_8);
byte[] compressed = lz4.compress(srcBytes, level);
byte[] decompressed = lz4.decompress(compressed);
// compression ratio may be negative for some random string data
// assertThat(compressed.length).isLessThan(srcBytes.length);
assertThat(decompressed).isEqualTo(srcBytes);
assertThat(new String(decompressed)).isEqualTo(message);
}
}
@Test(expected = RuntimeException.class)
public void testCompressionUnsupportedType() {
CompressionType.of("snappy");
}
}
common/src/test/java/org/apache/rocketmq/common/sysflag/CompressionFlagTest.java
+55
View File
@@ -0,0 +1,55 @@
/*
* 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.sysflag;
import org.apache.rocketmq.common.compression.CompressionType;
import org.junit.Test;
import static org.assertj.core.api.Assertions.assertThat;
public class CompressionFlagTest {
@Test
public void testCompressionFlag() {
int flag = 0;
flag |= MessageSysFlag.COMPRESSED_FLAG;
assertThat(MessageSysFlag.getCompressionType(flag)).isEqualTo(CompressionType.ZLIB);
flag |= MessageSysFlag.COMPRESSION_LZ4_TYPE;
assertThat(MessageSysFlag.getCompressionType(flag)).isEqualTo(CompressionType.LZ4);
flag &= (~MessageSysFlag.COMPRESSION_TYPE_COMPARATOR);
flag |= MessageSysFlag.COMPRESSION_ZSTD_TYPE;
assertThat(MessageSysFlag.getCompressionType(flag)).isEqualTo(CompressionType.ZSTD);
flag &= (~MessageSysFlag.COMPRESSION_TYPE_COMPARATOR);
flag |= MessageSysFlag.COMPRESSION_ZLIB_TYPE;
assertThat(MessageSysFlag.getCompressionType(flag)).isEqualTo(CompressionType.ZLIB);
}
@Test(expected = RuntimeException.class)
public void testCompressionFlagNotMatch() {
int flag = 0;
flag |= MessageSysFlag.COMPRESSED_FLAG;
flag |= (0x4 << 8);
MessageSysFlag.getCompressionType(flag);
}
}
pom.xml
+10 -2
View File
@@ -605,8 +605,16 @@
<artifactId>commons-validator</artifactId>
<version>1.7</version>
</dependency>
<dependency>
<groupId>com.github.luben</groupId>
<artifactId>zstd-jni</artifactId>
<version>1.5.2-2</version>
</dependency>
<dependency>
<groupId>org.lz4</groupId>
<artifactId>lz4-java</artifactId>
<version>1.8.0</version>
</dependency>
</dependencies>
</dependencyManagement>
</project>