DefaultMessageStore implements MessageStore { public GetMessageResult getMessage(final String group, final String topic, final int queueId, final long offset, final int maxMsgNums, final MessageFilter messageFilter) { if (this.shutdown) { log.warn("message store has shutdown, so getMessage is forbidden"); return null; } if (!this.runningFlags.isReadable()) { log.warn("message store is not readable, so getMessage is forbidden " + this.runningFlags.getFlagBits()); return null; } long beginTime = this.getSystemClock().now(); GetMessageStatus status = GetMessageStatus.NO_MESSAGE_IN_QUEUE; long nextBeginOffset = offset; long minOffset = 0; long maxOffset = 0; GetMessageResult getResult = new GetMessageResult(); final long maxOffsetPy = this.commitLog.getMaxOffset(); ConsumeQueue consumeQueue = findConsumeQueue(topic, queueId); if (consumeQueue != null) { minOffset = consumeQueue.getMinOffsetInQueue(); maxOffset = consumeQueue.getMaxOffsetInQueue(); if (maxOffset == 0) { status = GetMessageStatus.NO_MESSAGE_IN_QUEUE; nextBeginOffset = nextOffsetCorrection(offset, 0); } else if (offset < minOffset) { status = GetMessageStatus.OFFSET_TOO_SMALL; nextBeginOffset = nextOffsetCorrection(offset, minOffset); } else if (offset == maxOffset) { status = GetMessageStatus.OFFSET_OVERFLOW_ONE; nextBeginOffset = nextOffsetCorrection(offset, offset); } else if (offset > maxOffset) { status = GetMessageStatus.OFFSET_OVERFLOW_BADLY; if (0 == minOffset) { nextBeginOffset = nextOffsetCorrection(offset, minOffset); } else { nextBeginOffset = nextOffsetCorrection(offset, maxOffset); } } else { SelectMappedBufferResult bufferConsumeQueue = consumeQueue.getIndexBuffer(offset); if (bufferConsumeQueue != null) { try { status = GetMessageStatus.NO_MATCHED_MESSAGE; long nextPhyFileStartOffset = Long.MIN_VALUE; long maxPhyOffsetPulling = 0; int i = 0; final int maxFilterMessageCount = Math.max(16000, maxMsgNums * ConsumeQueue.CQ_STORE_UNIT_SIZE); final boolean diskFallRecorded = this.messageStoreConfig.isDiskFallRecorded(); ConsumeQueueExt.CqExtUnit cqExtUnit = new ConsumeQueueExt.CqExtUnit(); for (; i < bufferConsumeQueue.getSize() && i < maxFilterMessageCount; i += ConsumeQueue.CQ_STORE_UNIT_SIZE) { long offsetPy = bufferConsumeQueue.getByteBuffer().getLong(); int sizePy = bufferConsumeQueue.getByteBuffer().getInt(); long tagsCode = bufferConsumeQueue.getByteBuffer().getLong(); maxPhyOffsetPulling = offsetPy; if (nextPhyFileStartOffset != Long.MIN_VALUE) { if (offsetPy < nextPhyFileStartOffset) continue; } boolean isInDisk = checkInDiskByCommitOffset(offsetPy, maxOffsetPy); if (this.isTheBatchFull(sizePy, maxMsgNums, getResult.getBufferTotalSize(), getResult.getMessageCount(), isInDisk)) { break; } boolean extRet = false, isTagsCodeLegal = true; if (consumeQueue.isExtAddr(tagsCode)) { extRet = consumeQueue.getExt(tagsCode, cqExtUnit); if (extRet) { tagsCode = cqExtUnit.getTagsCode(); } else { log.error("[BUG] can't find consume queue extend file content!addr={}, offsetPy={}, sizePy={}, topic={}, group={}", tagsCode, offsetPy, sizePy, topic, group); isTagsCodeLegal = false; } } if (messageFilter != null && !messageFilter.isMatchedByConsumeQueue(isTagsCodeLegal ? tagsCode : null, extRet ? cqExtUnit : null)) { if (getResult.getBufferTotalSize() == 0) { status = GetMessageStatus.NO_MATCHED_MESSAGE; } continue; } SelectMappedBufferResult selectResult = this.commitLog.getMessage(offsetPy, sizePy); if (null == selectResult) { if (getResult.getBufferTotalSize() == 0) { status = GetMessageStatus.MESSAGE_WAS_REMOVING; } nextPhyFileStartOffset = this.commitLog.rollNextFile(offsetPy); continue; } if (messageFilter != null && !messageFilter.isMatchedByCommitLog(selectResult.getByteBuffer().slice(), null)) { if (getResult.getBufferTotalSize() == 0) { status = GetMessageStatus.NO_MATCHED_MESSAGE; } selectResult.release(); continue; } this.storeStatsService.getGetMessageTransferedMsgCount().incrementAndGet(); getResult.addMessage(selectResult); status = GetMessageStatus.FOUND; nextPhyFileStartOffset = Long.MIN_VALUE; } if (diskFallRecorded) { long fallBehind = maxOffsetPy - maxPhyOffsetPulling; brokerStatsManager.recordDiskFallBehindSize(group, topic, queueId, fallBehind); } nextBeginOffset = offset + (i / ConsumeQueue.CQ_STORE_UNIT_SIZE); long diff = maxOffsetPy - maxPhyOffsetPulling; long memory = (long) (StoreUtil.TOTAL_PHYSICAL_MEMORY_SIZE * (this.messageStoreConfig.getAccessMessageInMemoryMaxRatio() / 100.0)); getResult.setSuggestPullingFromSlave(diff > memory); } finally { bufferConsumeQueue.release(); } } else { status = GetMessageStatus.OFFSET_FOUND_NULL; nextBeginOffset = nextOffsetCorrection(offset, consumeQueue.rollNextFile(offset)); log.warn("consumer request topic: " + topic + "offset: " + offset + " minOffset: " + minOffset + " maxOffset: " + maxOffset + ", but access logic queue failed."); } } } else { status = GetMessageStatus.NO_MATCHED_LOGIC_QUEUE; nextBeginOffset = nextOffsetCorrection(offset, 0); } if (GetMessageStatus.FOUND == status) { this.storeStatsService.getGetMessageTimesTotalFound().incrementAndGet(); } else { this.storeStatsService.getGetMessageTimesTotalMiss().incrementAndGet(); } long eclipseTime = this.getSystemClock().now() - beginTime; this.storeStatsService.setGetMessageEntireTimeMax(eclipseTime); getResult.setStatus(status); getResult.setNextBeginOffset(nextBeginOffset); getResult.setMaxOffset(maxOffset); getResult.setMinOffset(minOffset); return getResult; } DefaultMessageStore(final MessageStoreConfig messageStoreConfig, final BrokerStatsManager brokerStatsManager,
        final MessageArrivingListener messageArrivingListener, final BrokerConfig brokerConfig); void truncateDirtyLogicFiles(long phyOffset); boolean load(); void start(); void shutdown(); void destroy(); void destroyLogics(); PutMessageResult putMessage(MessageExtBrokerInner msg); PutMessageResult putMessages(MessageExtBatch messageExtBatch); @Override boolean isOSPageCacheBusy(); @Override long lockTimeMills(); SystemClock getSystemClock(); CommitLog getCommitLog(); GetMessageResult getMessage(final String group, final String topic, final int queueId, final long offset,
        final int maxMsgNums,
        final MessageFilter messageFilter); long getMaxOffsetInQueue(String topic, int queueId); long getMinOffsetInQueue(String topic, int queueId); @Override long getCommitLogOffsetInQueue(String topic, int queueId, long consumeQueueOffset); long getOffsetInQueueByTime(String topic, int queueId, long timestamp); MessageExt lookMessageByOffset(long commitLogOffset); @Override SelectMappedBufferResult selectOneMessageByOffset(long commitLogOffset); @Override SelectMappedBufferResult selectOneMessageByOffset(long commitLogOffset, int msgSize); String getRunningDataInfo(); @Override HashMap<String, String> getRuntimeInfo(); @Override long getMaxPhyOffset(); @Override long getMinPhyOffset(); @Override long getEarliestMessageTime(String topic, int queueId); @Override long getEarliestMessageTime(); @Override long getMessageStoreTimeStamp(String topic, int queueId, long consumeQueueOffset); @Override long getMessageTotalInQueue(String topic, int queueId); @Override SelectMappedBufferResult getCommitLogData(final long offset); @Override boolean appendToCommitLog(long startOffset, byte[] data); @Override void executeDeleteFilesManually(); @Override QueryMessageResult queryMessage(String topic, String key, int maxNum, long begin, long end); @Override void updateHaMasterAddress(String newAddr); @Override long slaveFallBehindMuch(); @Override long now(); @Override int cleanUnusedTopic(Set<String> topics); void cleanExpiredConsumerQueue(); Map<String, Long> getMessageIds(final String topic, final int queueId, long minOffset, long maxOffset,
        SocketAddress storeHost); @Override boolean checkInDiskByConsumeOffset(final String topic, final int queueId, long consumeOffset); long dispatchBehindBytes(); @Override long flush(); @Override boolean resetWriteOffset(long phyOffset); @Override long getConfirmOffset(); @Override void setConfirmOffset(long phyOffset); MessageExt lookMessageByOffset(long commitLogOffset, int size); ConsumeQueue findConsumeQueue(String topic, int queueId); MessageStoreConfig getMessageStoreConfig(); TransientStorePool getTransientStorePool(); AllocateMappedFileService getAllocateMappedFileService(); StoreStatsService getStoreStatsService(); RunningFlags getAccessRights(); ConcurrentMap<String, ConcurrentMap<Integer, ConsumeQueue>> getConsumeQueueTable(); StoreCheckpoint getStoreCheckpoint(); HAService getHaService(); ScheduleMessageService getScheduleMessageService(); RunningFlags getRunningFlags(); void doDispatch(DispatchRequest req); void putMessagePositionInfo(DispatchRequest dispatchRequest); BrokerStatsManager getBrokerStatsManager(); int remainTransientStoreBufferNumbs(); @Override boolean isTransientStorePoolDeficient(); @Override LinkedList<CommitLogDispatcher> getDispatcherList(); @Override ConsumeQueue getConsumeQueue(String topic, int queueId); void unlockMappedFile(final MappedFile mappedFile);  }
@Test public void testReadScheduleMsg() throws Exception { System.err.println(System.currentTimeMillis()); long totalMsgs = 6000; QUEUE_TOTAL = 10; MessageBody = StoreMessage.getBytes(); int offset=0; for (long i = 0; i < totalMsgs; i++) { Thread.sleep(1000); GetMessageResult result = messageStore.getMessage("GROUP_B", "FooBar", 0, offset, 1024 * 1024, null); System.err.println(offset); offset+=result.getMessageCount(); List<ByteBuffer> ll=result.getMessageBufferList(); if(ll!=null) { } result.release(); } }