opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-09-24 16:04:00 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #5837] Deprecate MessageStore#checkInDiskByConsumeOffset (#5840)

Browse Source 
* deprecate MessageStore#checkInDiskByConsumeOffset and add checkInMemByConsumeOffset and checkInStoreByConsumeOffset in MessageStore

* skip initializing isLoaded0 method in windows
This commit is contained in:
SSpirits
2023-01-09 16:47:42 +08:00
committed by GitHub
parent 98ebcef71c
commit 3949f6938e
9 changed files with 176 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/apache/rocketmq/broker/offset/BroadcastOffsetManager.java
+1 -1
View File
@@ -116,7 +116,7 @@ public class BroadcastOffsetManager extends ServiceThread {
brokerController.getConsumerOffsetManager().queryOffset(broadcastGroupId(groupId), topic, queueId);
}
if (storeOffset < 0) {
if (!this.brokerController.getMessageStore().checkInDiskByConsumeOffset(topic, queueId, 0)) {
if (this.brokerController.getMessageStore().checkInMemByConsumeOffset(topic, queueId, 0, 1)) {
storeOffset = 0;
} else {
storeOffset = brokerController.getMessageStore().getMaxOffsetInQueue(topic, queueId, true);
broker/src/main/java/org/apache/rocketmq/broker/processor/ConsumerManageProcessor.java
+2 -2
View File
@@ -328,8 +328,8 @@ public class ConsumerManageProcessor implements NettyRequestProcessor {
response.setCode(ResponseCode.QUERY_NOT_FOUND);
response.setRemark("Not found, do not set to zero, maybe this group boot first");
} else if (minOffset <= 0
&& !this.brokerController.getMessageStore().checkInDiskByConsumeOffset(
requestHeader.getTopic(), requestHeader.getQueueId(), 0)) {
&& this.brokerController.getMessageStore().checkInMemByConsumeOffset(
requestHeader.getTopic(), requestHeader.getQueueId(), 0, 1)) {
responseHeader.setOffset(0L);
response.setCode(ResponseCode.SUCCESS);
response.setRemark(null);
store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java
+61 -18
View File
@@ -795,13 +795,13 @@ public class DefaultMessageStore implements MessageStore {
long offsetPy = cqUnit.getPos();
int sizePy = cqUnit.getSize();
boolean isInDisk = checkInDiskByCommitOffset(offsetPy, maxOffsetPy);
boolean isInMem = estimateInMemByCommitOffset(offsetPy, maxOffsetPy);
if ((cqUnit.getQueueOffset() - offset) * consumeQueue.getUnitSize() > maxFilterMessageSize) {
break;
}
if (this.isTheBatchFull(sizePy, cqUnit.getBatchNum(), maxMsgNums, maxPullSize, getResult.getBufferTotalSize(), getResult.getMessageCount(), isInDisk)) {
if (this.isTheBatchFull(sizePy, cqUnit.getBatchNum(), maxMsgNums, maxPullSize, getResult.getBufferTotalSize(), getResult.getMessageCount(), isInMem)) {
break;
}
@@ -1418,6 +1418,7 @@ public class DefaultMessageStore implements MessageStore {
}
@Override
@Deprecated
public boolean checkInDiskByConsumeOffset(final String topic, final int queueId, long consumeOffset) {
final long maxOffsetPy = this.commitLog.getMaxOffset();
@@ -1428,7 +1429,7 @@ public class DefaultMessageStore implements MessageStore {
if (cqUnit != null) {
long offsetPy = cqUnit.getPos();
return checkInDiskByCommitOffset(offsetPy, maxOffsetPy);
return !estimateInMemByCommitOffset(offsetPy, maxOffsetPy);
} else {
return false;
}
@@ -1436,6 +1437,38 @@ public class DefaultMessageStore implements MessageStore {
return false;
}
@Override
public boolean checkInMemByConsumeOffset(final String topic, final int queueId, long consumeOffset, int batchSize) {
ConsumeQueueInterface consumeQueue = findConsumeQueue(topic, queueId);
if (consumeQueue != null) {
CqUnit firstCQItem = consumeQueue.get(consumeOffset);
if (firstCQItem == null) {
return false;
}
long startOffsetPy = firstCQItem.getPos();
if (batchSize <= 1) {
int size = firstCQItem.getSize();
return checkInMemByCommitOffset(startOffsetPy, size);
}
CqUnit lastCQItem = consumeQueue.get(consumeOffset + batchSize);
if (lastCQItem == null) {
int size = firstCQItem.getSize();
return checkInMemByCommitOffset(startOffsetPy, size);
}
long endOffsetPy = lastCQItem.getPos();
int size = (int) (endOffsetPy - startOffsetPy) + lastCQItem.getSize();
return checkInMemByCommitOffset(startOffsetPy, size);
}
return false;
}
@Override
public boolean checkInStoreByConsumeOffset(String topic, int queueId, long consumeOffset) {
long commitLogOffset = getCommitLogOffsetInQueue(topic, queueId, consumeOffset);
return checkInDiskByCommitOffset(commitLogOffset);
}
@Override
public long dispatchBehindBytes() {
return this.reputMessageService.behind();
@@ -1600,13 +1633,29 @@ public class DefaultMessageStore implements MessageStore {
return nextOffset;
}
private boolean checkInDiskByCommitOffset(long offsetPy, long maxOffsetPy) {
private boolean estimateInMemByCommitOffset(long offsetPy, long maxOffsetPy) {
long memory = (long) (StoreUtil.TOTAL_PHYSICAL_MEMORY_SIZE * (this.messageStoreConfig.getAccessMessageInMemoryMaxRatio() / 100.0));
return (maxOffsetPy - offsetPy) > memory;
return (maxOffsetPy - offsetPy) <= memory;
}
private boolean checkInMemByCommitOffset(long offsetPy, int size) {
SelectMappedBufferResult message = this.commitLog.getMessage(offsetPy, size);
if (message != null) {
try {
return message.isInMem();
} finally {
message.release();
}
}
return false;
}
public boolean checkInDiskByCommitOffset(long offsetPy) {
return offsetPy >= commitLog.getMinOffset();
}
private boolean isTheBatchFull(int sizePy, int unitBatchNum, int maxMsgNums, long maxMsgSize, int bufferTotal,
int messageTotal, boolean isInDisk) {
int messageTotal, boolean isInMem) {
if (0 == bufferTotal || 0 == messageTotal) {
return false;
@@ -1620,25 +1669,19 @@ public class DefaultMessageStore implements MessageStore {
return true;
}
if (isInDisk) {
if ((bufferTotal + sizePy) > this.messageStoreConfig.getMaxTransferBytesOnMessageInDisk()) {
return true;
}
if (messageTotal > this.messageStoreConfig.getMaxTransferCountOnMessageInDisk() - 1) {
return true;
}
} else {
if (isInMem) {
if ((bufferTotal + sizePy) > this.messageStoreConfig.getMaxTransferBytesOnMessageInMemory()) {
return true;
}
if (messageTotal > this.messageStoreConfig.getMaxTransferCountOnMessageInMemory() - 1) {
return messageTotal > this.messageStoreConfig.getMaxTransferCountOnMessageInMemory() - 1;
} else {
if ((bufferTotal + sizePy) > this.messageStoreConfig.getMaxTransferBytesOnMessageInDisk()) {
return true;
}
}
return false;
return messageTotal > this.messageStoreConfig.getMaxTransferCountOnMessageInDisk() - 1;
}
}
private void deleteFile(final String fileName) {
store/src/main/java/org/apache/rocketmq/store/MessageStore.java
+27 -5
View File
@@ -129,7 +129,7 @@ public interface MessageStore {
/**
* Asynchronous get message
* @see org.apache.rocketmq.store.MessageStore#getMessage(String, String, int, long, int, MessageFilter) getMessage
* @see #getMessage(String, String, int, long, int, MessageFilter) getMessage
*
* @param group Consumer group that launches this query.
* @param topic Topic to query.
@@ -160,7 +160,7 @@ public interface MessageStore {
/**
* Asynchronous get message
* @see org.apache.rocketmq.store.MessageStore#getMessage(String, String, int, long, int, int, MessageFilter) getMessage
* @see #getMessage(String, String, int, long, int, int, MessageFilter) getMessage
*
* @param group Consumer group that launches this query.
* @param topic Topic to query.
@@ -312,7 +312,7 @@ public interface MessageStore {
/**
* Asynchronous get the store time of the earliest message in this store.
* @see org.apache.rocketmq.store.MessageStore#getEarliestMessageTime() getEarliestMessageTime
* @see #getEarliestMessageTime() getEarliestMessageTime
*
* @return timestamp of the earliest message in this store.
*/
@@ -330,7 +330,7 @@ public interface MessageStore {
/**
* Asynchronous get the store time of the message specified.
* @see org.apache.rocketmq.store.MessageStore#getMessageStoreTimeStamp(String, int, long) getMessageStoreTimeStamp
* @see #getMessageStoreTimeStamp(String, int, long) getMessageStoreTimeStamp
*
* @param topic message topic.
* @param queueId queue ID.
@@ -396,7 +396,7 @@ public interface MessageStore {
/**
* Asynchronous query messages by given key.
* @see org.apache.rocketmq.store.MessageStore#queryMessage(String, String, int, long, long) queryMessage
* @see #queryMessage(String, String, int, long, long) queryMessage
*
* @param topic topic of the message.
* @param key message key.
@@ -464,9 +464,31 @@ public interface MessageStore {
* @param queueId queue ID.
* @param consumeOffset consume queue offset.
* @return true if the message is no longer in memory; false otherwise.
* @deprecated As of RIP-57, replaced by {@link #checkInMemByConsumeOffset(String, int, long, int)}, see <a href="https://github.com/apache/rocketmq/issues/5837">this issue</a> for more details
*/
@Deprecated
boolean checkInDiskByConsumeOffset(final String topic, final int queueId, long consumeOffset);
/**
* Check if the given message is in the page cache.
*
* @param topic topic.
* @param queueId queue ID.
* @param consumeOffset consume queue offset.
* @return true if the message is in page cache; false otherwise.
*/
boolean checkInMemByConsumeOffset(final String topic, final int queueId, long consumeOffset, int batchSize);
/**
* Check if the given message is in store.
*
* @param topic topic.
* @param queueId queue ID.
* @param consumeOffset consume queue offset.
* @return true if the message is in store; false otherwise.
*/
boolean checkInStoreByConsumeOffset(final String topic, final int queueId, long consumeOffset);
/**
* Get number of the bytes that have been stored in commit log and not yet dispatched to consume queue.
*
store/src/main/java/org/apache/rocketmq/store/SelectMappedBufferResult.java
+9 -2
View File
@@ -16,9 +16,8 @@
*/
package org.apache.rocketmq.store;
import org.apache.rocketmq.store.logfile.MappedFile;
import java.nio.ByteBuffer;
import org.apache.rocketmq.store.logfile.MappedFile;
public class SelectMappedBufferResult {
@@ -67,4 +66,12 @@ public class SelectMappedBufferResult {
public long getStartOffset() {
return startOffset;
}
public boolean isInMem() {
if (mappedFile == null) {
return true;
}
long pos = startOffset - mappedFile.getFileFromOffset();
return mappedFile.isLoaded(pos, size);
}
}
store/src/main/java/org/apache/rocketmq/store/logfile/DefaultMappedFile.java
+55
View File
@@ -22,6 +22,8 @@ import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.nio.ByteBuffer;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
@@ -35,6 +37,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;
import org.apache.commons.lang3.SystemUtils;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.message.MessageExt;
@@ -52,10 +55,15 @@ import org.apache.rocketmq.store.SelectMappedBufferResult;
import org.apache.rocketmq.store.TransientStorePool;
import org.apache.rocketmq.store.config.FlushDiskType;
import org.apache.rocketmq.store.util.LibC;
import sun.misc.Unsafe;
import sun.nio.ch.DirectBuffer;
public class DefaultMappedFile extends AbstractMappedFile {
public static final int OS_PAGE_SIZE = 1024 * 4;
public static final Unsafe UNSAFE = getUnsafe();
private static final Method IS_LOADED_METHOD;
public static final int UNSAFE_PAGE_SIZE = UNSAFE == null ? OS_PAGE_SIZE : UNSAFE.pageSize();
protected static final Logger log = LoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME);
protected static final AtomicLong TOTAL_MAPPED_VIRTUAL_MEMORY = new AtomicLong(0);
@@ -92,6 +100,18 @@ public class DefaultMappedFile extends AbstractMappedFile {
WROTE_POSITION_UPDATER = AtomicIntegerFieldUpdater.newUpdater(DefaultMappedFile.class, "wrotePosition");
COMMITTED_POSITION_UPDATER = AtomicIntegerFieldUpdater.newUpdater(DefaultMappedFile.class, "committedPosition");
FLUSHED_POSITION_UPDATER = AtomicIntegerFieldUpdater.newUpdater(DefaultMappedFile.class, "flushedPosition");
Method isLoaded0method = null;
// On the windows platform and openjdk 11 method isLoaded0 always returns false.
// see https://github.com/AdoptOpenJDK/openjdk-jdk11/blob/19fb8f93c59dfd791f62d41f332db9e306bc1422/src/java.base/windows/native/libnio/MappedByteBuffer.c#L34
if (!SystemUtils.IS_OS_WINDOWS) {
try {
isLoaded0method = MappedByteBuffer.class.getDeclaredMethod("isLoaded0", long.class, long.class, int.class);
isLoaded0method.setAccessible(true);
} catch (NoSuchMethodException ignore) {
}
}
IS_LOADED_METHOD = isLoaded0method;
}
public DefaultMappedFile() {
@@ -796,6 +816,41 @@ public class DefaultMappedFile extends AbstractMappedFile {
return new Itr(startPos);
}
public static Unsafe getUnsafe() {
try {
Field f = Unsafe.class.getDeclaredField("theUnsafe");
f.setAccessible(true);
return (Unsafe) f.get(null);
} catch (Exception ignore) {
}
return null;
}
public static long mappingAddr(long addr) {
long offset = addr % UNSAFE_PAGE_SIZE;
offset = (offset >= 0) ? offset : (UNSAFE_PAGE_SIZE + offset);
return addr - offset;
}
public static int pageCount(long size) {
return (int) (size + (long) UNSAFE_PAGE_SIZE - 1L) / UNSAFE_PAGE_SIZE;
}
@Override
public boolean isLoaded(long position, int size) {
if (IS_LOADED_METHOD == null) {
return true;
}
try {
long addr = ((DirectBuffer) mappedByteBuffer).address() + position;
return (boolean) IS_LOADED_METHOD.invoke(mappedByteBuffer, mappingAddr(addr), size, pageCount(size));
} catch (Exception e) {
log.info("invoke isLoaded0 of file {} error:", file.getAbsolutePath(), e);
}
return true;
}
private class Itr implements Iterator<SelectMappedBufferResult> {
private int start;
private int current;
store/src/main/java/org/apache/rocketmq/store/logfile/MappedFile.java
+8
View File
@@ -360,4 +360,12 @@ public interface MappedFile {
void init(String fileName, int fileSize, TransientStorePool transientStorePool) throws IOException;
Iterator<SelectMappedBufferResult> iterator(int pos);
/**
* Check mapped file is loaded to memory with given position and size
* @param position start offset of data
* @param size data size
* @return data is resided in memory or not
*/
boolean isLoaded(long position, int size);
}
store/src/main/java/org/apache/rocketmq/store/plugin/AbstractPluginMessageStore.java
+11
View File
@@ -272,10 +272,21 @@ public abstract class AbstractPluginMessageStore implements MessageStore {
}
@Override
@Deprecated
public boolean checkInDiskByConsumeOffset(String topic, int queueId, long consumeOffset) {
return next.checkInDiskByConsumeOffset(topic, queueId, consumeOffset);
}
@Override
public boolean checkInMemByConsumeOffset(String topic, int queueId, long consumeOffset, int batchSize) {
return next.checkInMemByConsumeOffset(topic, queueId, consumeOffset, batchSize);
}
@Override
public boolean checkInStoreByConsumeOffset(String topic, int queueId, long consumeOffset) {
return next.checkInStoreByConsumeOffset(topic, queueId, consumeOffset);
}
@Override
public long dispatchBehindBytes() {
return next.dispatchBehindBytes();
store/src/test/java/org/apache/rocketmq/store/queue/BatchConsumeMessageTest.java
+2 -2
View File
@@ -281,7 +281,7 @@ public class BatchConsumeMessageTest extends QueueTestBase {
Assert.assertTrue(commitLogOffset <= messageStore.getMaxPhyOffset());
Assert.assertEquals(commitLogMid, commitLogOffset);
Assert.assertFalse(messageStore.checkInDiskByConsumeOffset(topic, 0, 50));
Assert.assertTrue(messageStore.checkInMemByConsumeOffset(topic, 0, 50, 1));
}
@Test
@@ -331,7 +331,7 @@ public class BatchConsumeMessageTest extends QueueTestBase {
Assert.assertTrue(commitLogOffset >= messageStore.getMinPhyOffset());
Assert.assertTrue(commitLogOffset <= messageStore.getMaxPhyOffset());
Assert.assertFalse(messageStore.checkInDiskByConsumeOffset(topic, 0, 300));
Assert.assertTrue(messageStore.checkInMemByConsumeOffset(topic, 0, 300, 1));
//get the message Normally
GetMessageResult getMessageResult = messageStore.getMessage("group", topic, 0, 0, 10 * batchNum, null);