Skip to content

Commit 25c4e1c

Browse files
RongtongJinRongtongJin
committed
Merge branch master into develop
2 parents 881f767 + c8892e1 commit 25c4e1c

File tree

3 files changed

+45
-20
lines changed

3 files changed

+45
-20
lines changed

‎README.md‎

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -11,11 +11,12 @@
1111

1212
It offers a variety of features:
1313

14-
* Pub/Sub messaging model
14+
* Messageing patterns including publish/subscribe, request/reply and streaming
1515
* Financial grade transactional message
16+
* Built-in fault tolerance and high availability configuration options base on [DLedger](https://github.com/openmessaging/openmessaging-storage-dledger)
1617
* A variety of cross language clients, such as Java, C/C++, Python, Go
1718
* Pluggable transport protocols, such as TCP, SSL, AIO
18-
* Inbuilt message tracing capability, also support opentracing
19+
* Built-in message tracing capability, also support opentracing
1920
* Versatile big-data and streaming ecosytem integration
2021
* Message retroactivity by time or offset
2122
* Reliable FIFO and strict ordered messaging in the same queue
@@ -27,7 +28,8 @@ It offers a variety of features:
2728
* Various message filter mechanics such as SQL and Tag
2829
* Docker images for isolated testing and cloud isolated clusters
2930
* Feature-rich administrative dashboard for configuration, metrics and monitoring
30-
* Authentication and authorisation
31+
* Authentication and authorization
32+
* Free open source connectors, for both sources and sinks
3133

3234
----------
3335

‎example/src/main/java/org/apache/rocketmq/example/benchmark/Consumer.java‎

Lines changed: 32 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,9 @@
2222
import java.util.List;
2323
import java.util.Timer;
2424
import java.util.TimerTask;
25+
import java.util.concurrent.ThreadLocalRandom;
2526
import java.util.concurrent.atomic.AtomicLong;
27+
2628
import org.apache.commons.cli.CommandLine;
2729
import org.apache.commons.cli.Option;
2830
import org.apache.commons.cli.Options;
@@ -49,15 +51,16 @@ public static void main(String[] args) throws MQClientException, IOException {
4951

5052
final String topic = commandLine.hasOption('t') ? commandLine.getOptionValue('t').trim() : "BenchmarkTest";
5153
final String groupPrefix = commandLine.hasOption('g') ? commandLine.getOptionValue('g').trim() : "benchmark_consumer";
52-
final String isPrefixEnable = commandLine.hasOption('p') ? commandLine.getOptionValue('p').trim() : "true";
54+
final String isSuffixEnable = commandLine.hasOption('p') ? commandLine.getOptionValue('p').trim() : "true";
5355
final String filterType = commandLine.hasOption('f') ? commandLine.getOptionValue('f').trim() : null;
5456
final String expression = commandLine.hasOption('e') ? commandLine.getOptionValue('e').trim() : null;
57+
final double failRate = commandLine.hasOption('r') ? Double.parseDouble(commandLine.getOptionValue('r').trim()) : 0.0;
5558
String group = groupPrefix;
56-
if (Boolean.parseBoolean(isPrefixEnable)) {
57-
group = groupPrefix + "_" + Long.toString(System.currentTimeMillis() % 100);
59+
if (Boolean.parseBoolean(isSuffixEnable)) {
60+
group = groupPrefix + "_" + (System.currentTimeMillis() % 100);
5861
}
5962

60-
System.out.printf("topic: %s, group: %s, prefix: %s, filterType: %s, expression: %s%n", topic, group, isPrefixEnable, filterType, expression);
63+
System.out.printf("topic: %s, group: %s, suffix: %s, filterType: %s, expression: %s%n", topic, group, isSuffixEnable, filterType, expression);
6164

6265
final StatsBenchmarkConsumer statsBenchmarkConsumer = new StatsBenchmarkConsumer();
6366

@@ -85,9 +88,15 @@ private void printStats() {
8588
(long) (((end[1] - begin[1]) / (double) (end[0] - begin[0])) * 1000L);
8689
final double averageB2CRT = (end[2] - begin[2]) / (double) (end[1] - begin[1]);
8790
final double averageS2CRT = (end[3] - begin[3]) / (double) (end[1] - begin[1]);
91+
final long failCount = end[4] - begin[4];
92+
final long b2cMax = statsBenchmarkConsumer.getBorn2ConsumerMaxRT().get();
93+
final long s2cMax = statsBenchmarkConsumer.getStore2ConsumerMaxRT().get();
94+
95+
statsBenchmarkConsumer.getBorn2ConsumerMaxRT().set(0);
96+
statsBenchmarkConsumer.getStore2ConsumerMaxRT().set(0);
8897

89-
System.out.printf("Consume TPS: %d Average(B2C) RT: %7.3f Average(S2C) RT: %7.3f MAX(B2C) RT: %d MAX(S2C) RT: %d%n",
90-
consumeTps, averageB2CRT, averageS2CRT, end[4], end[5]
98+
System.out.printf("TPS: %d FAIL: %d AVG(B2C) RT: %7.3f AVG(S2C) RT: %7.3f MAX(B2C) RT: %d MAX(S2C) RT: %d%n",
99+
consumeTps, failCount, averageB2CRT, averageS2CRT, b2cMax, s2cMax
91100
);
92101
}
93102
}
@@ -144,7 +153,12 @@ public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
144153

145154
compareAndSetMax(statsBenchmarkConsumer.getStore2ConsumerMaxRT(), store2ConsumerRT);
146155

147-
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
156+
if (ThreadLocalRandom.current().nextDouble() < failRate) {
157+
statsBenchmarkConsumer.getFailCount().incrementAndGet();
158+
return ConsumeConcurrentlyStatus.RECONSUME_LATER;
159+
} else {
160+
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
161+
}
148162
}
149163
});
150164

@@ -174,6 +188,10 @@ public static Options buildCommandlineOptions(final Options options) {
174188
opt.setRequired(false);
175189
options.addOption(opt);
176190

191+
opt = new Option("r", "fail rate", true, "consumer fail rate, default 0");
192+
opt.setRequired(false);
193+
options.addOption(opt);
194+
177195
return options;
178196
}
179197

@@ -200,14 +218,15 @@ class StatsBenchmarkConsumer {
200218

201219
private final AtomicLong store2ConsumerMaxRT = new AtomicLong(0L);
202220

221+
private final AtomicLong failCount = new AtomicLong(0L);
222+
203223
public Long[] createSnapshot() {
204224
Long[] snap = new Long[] {
205225
System.currentTimeMillis(),
206226
this.receiveMessageTotalCount.get(),
207227
this.born2ConsumerTotalRT.get(),
208228
this.store2ConsumerTotalRT.get(),
209-
this.born2ConsumerMaxRT.get(),
210-
this.store2ConsumerMaxRT.get(),
229+
this.failCount.get()
211230
};
212231

213232
return snap;
@@ -232,4 +251,8 @@ public AtomicLong getBorn2ConsumerMaxRT() {
232251
public AtomicLong getStore2ConsumerMaxRT() {
233252
return store2ConsumerMaxRT;
234253
}
254+
255+
public AtomicLong getFailCount() {
256+
return failCount;
257+
}
235258
}

‎store/src/main/java/org/apache/rocketmq/store/CommitLog.java‎

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -596,7 +596,7 @@ public PutMessageResult putMessage(final MessageExtBrokerInner msg) {
596596
msg.setStoreHostAddressV6Flag();
597597
}
598598

599-
long eclipsedTimeInLock = 0;
599+
long elapsedTimeInLock = 0;
600600

601601
MappedFile unlockMappedFile = null;
602602
MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile();
@@ -647,14 +647,14 @@ public PutMessageResult putMessage(final MessageExtBrokerInner msg) {
647647
return new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result);
648648
}
649649

650-
eclipsedTimeInLock = this.defaultMessageStore.getSystemClock().now() - beginLockTimestamp;
650+
elapsedTimeInLock = this.defaultMessageStore.getSystemClock().now() - beginLockTimestamp;
651651
beginTimeInLock = 0;
652652
} finally {
653653
putMessageLock.unlock();
654654
}
655655

656-
if (eclipsedTimeInLock > 500) {
657-
log.warn("[NOTIFYME]putMessage in lock cost time(ms)={}, bodyLength={} AppendMessageResult={}", eclipsedTimeInLock, msg.getBody().length, result);
656+
if (elapsedTimeInLock > 500) {
657+
log.warn("[NOTIFYME]putMessage in lock cost time(ms)={}, bodyLength={} AppendMessageResult={}", elapsedTimeInLock, msg.getBody().length, result);
658658
}
659659

660660
if (null != unlockMappedFile && this.defaultMessageStore.getMessageStoreConfig().isWarmMapedFileEnable()) {
@@ -752,7 +752,7 @@ public PutMessageResult putMessages(final MessageExtBatch messageExtBatch) {
752752
messageExtBatch.setStoreHostAddressV6Flag();
753753
}
754754

755-
long eclipsedTimeInLock = 0;
755+
long elapsedTimeInLock = 0;
756756
MappedFile unlockMappedFile = null;
757757
MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile();
758758

@@ -807,14 +807,14 @@ public PutMessageResult putMessages(final MessageExtBatch messageExtBatch) {
807807
return new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result);
808808
}
809809

810-
eclipsedTimeInLock = this.defaultMessageStore.getSystemClock().now() - beginLockTimestamp;
810+
elapsedTimeInLock = this.defaultMessageStore.getSystemClock().now() - beginLockTimestamp;
811811
beginTimeInLock = 0;
812812
} finally {
813813
putMessageLock.unlock();
814814
}
815815

816-
if (eclipsedTimeInLock > 500) {
817-
log.warn("[NOTIFYME]putMessages in lock cost time(ms)={}, bodyLength={} AppendMessageResult={}", eclipsedTimeInLock, messageExtBatch.getBody().length, result);
816+
if (elapsedTimeInLock > 500) {
817+
log.warn("[NOTIFYME]putMessages in lock cost time(ms)={}, bodyLength={} AppendMessageResult={}", elapsedTimeInLock, messageExtBatch.getBody().length, result);
818818
}
819819

820820
if (null != unlockMappedFile && this.defaultMessageStore.getMessageStoreConfig().isWarmMapedFileEnable()) {

0 commit comments

Comments
 (0)