diff --git a/store/src/main/java/org/apache/rocketmq/store/AppendMessageResult.java b/store/src/main/java/org/apache/rocketmq/store/AppendMessageResult.java index 98bf203ad5d..fe5da43c19d 100644 --- a/store/src/main/java/org/apache/rocketmq/store/AppendMessageResult.java +++ b/store/src/main/java/org/apache/rocketmq/store/AppendMessageResult.java @@ -84,6 +84,18 @@ public AppendMessageResult(AppendMessageStatus status, long wroteOffset, int wro this.msgNum = msgNum; } + public AppendMessageResult(AppendMessageStatus status, long wroteOffset, int wroteBytes, String msgId, + long storeTimestamp, long logicsOffset, long pagecacheRT, int msgNum) { + this.status = status; + this.wroteOffset = wroteOffset; + this.wroteBytes = wroteBytes; + this.msgId = msgId; + this.storeTimestamp = storeTimestamp; + this.logicsOffset = logicsOffset; + this.pagecacheRT = pagecacheRT; + this.msgNum = msgNum; + } + public long getPagecacheRT() { return pagecacheRT; } diff --git a/store/src/main/java/org/apache/rocketmq/store/index/IndexService.java b/store/src/main/java/org/apache/rocketmq/store/index/IndexService.java index 1a180e4442b..8360329cb5a 100644 --- a/store/src/main/java/org/apache/rocketmq/store/index/IndexService.java +++ b/store/src/main/java/org/apache/rocketmq/store/index/IndexService.java @@ -49,6 +49,7 @@ public class IndexService implements CommitLogDispatchStore { private final String storePath; private final ArrayList indexFileList = new ArrayList<>(); private final ReadWriteLock readWriteLock = new ReentrantReadWriteLock(); + private static final ThreadLocal KEY_BUILDER = ThreadLocal.withInitial(() -> new StringBuilder(128)); public IndexService(final DefaultMessageStore store) { this.defaultMessageStore = store; @@ -215,10 +216,15 @@ public QueryOffsetResult queryOffset(String topic, String key, int maxNum, long } private String buildKey(final String topic, final String key) { - return topic + "#" + key; + StringBuilder keyBuilder = KEY_BUILDER.get(); + keyBuilder.setLength(0); + return keyBuilder.append(topic).append('#').append(key).toString(); } + private String buildKey(final String topic, final String key, final String indexType) { - return topic + "#" + indexType + "#" + key; + StringBuilder keyBuilder = KEY_BUILDER.get(); + keyBuilder.setLength(0); + return keyBuilder.append(topic).append('#').append(indexType).append('#').append(key).toString(); } public void buildIndex(DispatchRequest req) { diff --git a/store/src/main/java/org/apache/rocketmq/store/timer/TimerWheel.java b/store/src/main/java/org/apache/rocketmq/store/timer/TimerWheel.java index 2d5ce382012..c3498c986af 100644 --- a/store/src/main/java/org/apache/rocketmq/store/timer/TimerWheel.java +++ b/store/src/main/java/org/apache/rocketmq/store/timer/TimerWheel.java @@ -55,6 +55,8 @@ protected ByteBuffer initialValue() { }; private final int wheelLength; + private volatile boolean dirty; + private long snapOffset; public TimerWheel(String fileName, int slotsTotal, int precisionMs) throws IOException { @@ -128,14 +130,26 @@ public void flush() { if (mappedByteBuffer == null) { return; } + if (!dirty) { + return; + } + // Clear the flag before diffing: a concurrent putSlot during the diff re-marks it and the + // next flush picks the change up, so an interleaving can only cause an extra flush, never a + // missed one. All threads share the same underlying direct buffer (localBuffer duplicates + // only isolate position/limit), so the data itself is always visible here. + dirty = false; ByteBuffer bf = localBuffer.get(); - bf.position(0); - bf.limit(wheelLength); - mappedByteBuffer.position(0); - mappedByteBuffer.limit(wheelLength); - for (int i = 0; i < wheelLength; i++) { - if (bf.get(i) != mappedByteBuffer.get(i)) { - mappedByteBuffer.put(i, bf.get(i)); + int longAligned = wheelLength & ~7; + for (int i = 0; i < longAligned; i += 8) { + long local = bf.getLong(i); + if (local != mappedByteBuffer.getLong(i)) { + mappedByteBuffer.putLong(i, local); + } + } + for (int i = longAligned; i < wheelLength; i++) { + byte b = bf.get(i); + if (b != mappedByteBuffer.get(i)) { + mappedByteBuffer.put(i, b); } } this.mappedByteBuffer.force(); @@ -287,11 +301,10 @@ public int getSlotIndex(long timeMs) { public void putSlot(long timeMs, long firstPos, long lastPos) { localBuffer.get().position(getSlotIndex(timeMs) * Slot.SIZE); - // To be compatible with previous version. - // The previous version's precision is fixed at 1000ms and it store timeMs / 1000 in slot. localBuffer.get().putLong(timeMs / precisionMs); localBuffer.get().putLong(firstPos); localBuffer.get().putLong(lastPos); + dirty = true; } public void putSlot(long timeMs, long firstPos, long lastPos, int num, int magic) { @@ -301,6 +314,7 @@ public void putSlot(long timeMs, long firstPos, long lastPos, int num, int magic localBuffer.get().putLong(lastPos); localBuffer.get().putInt(num); localBuffer.get().putInt(magic); + dirty = true; } public void reviseSlot(long timeMs, long firstPos, long lastPos, boolean force) { @@ -313,11 +327,13 @@ public void reviseSlot(long timeMs, long firstPos, long lastPos, boolean force) } else { if (IGNORE != firstPos) { localBuffer.get().putLong(firstPos); + dirty = true; } else { localBuffer.get().getLong(); } if (IGNORE != lastPos) { localBuffer.get().putLong(lastPos); + dirty = true; } } } diff --git a/store/src/test/java/org/apache/rocketmq/store/stats/BrokerStatsManagerTest.java b/store/src/test/java/org/apache/rocketmq/store/stats/BrokerStatsManagerTest.java index 058ad0b0208..de091f6cae1 100644 --- a/store/src/test/java/org/apache/rocketmq/store/stats/BrokerStatsManagerTest.java +++ b/store/src/test/java/org/apache/rocketmq/store/stats/BrokerStatsManagerTest.java @@ -45,7 +45,7 @@ public class BrokerStatsManagerTest { private BrokerStatsManager brokerStatsManager; private static final String TOPIC = "TOPIC_TEST"; - private static final Integer QUEUE_ID = 0; + private static final int QUEUE_ID = 0; private static final String GROUP_NAME = "GROUP_TEST"; private static final String CLUSTER_NAME = "DefaultCluster";