@Test public void testReadOneScheduleMsg() throws Exception { long totalMsgs = 60000; QUEUE_TOTAL = 10; MessageExt msgExt = messageStore.lookMessageByOffset( 9677008); System.err.println(msgExt); System.err.println(System.currentTimeMillis()); }
public MessageExt lookMessageByOffset(long commitLogOffset) { SelectMappedBufferResult sbr = this.commitLog.getMessage(commitLogOffset, 4); if (null != sbr) { try { int size = sbr.getByteBuffer().getInt(); return lookMessageByOffset(commitLogOffset, size); } finally { sbr.release(); } } return null; }
DefaultMessageStore implements MessageStore { public MessageExt lookMessageByOffset(long commitLogOffset) { SelectMappedBufferResult sbr = this.commitLog.getMessage(commitLogOffset, 4); if (null != sbr) { try { int size = sbr.getByteBuffer().getInt(); return lookMessageByOffset(commitLogOffset, size); } finally { sbr.release(); } } return null; } }
DefaultMessageStore implements MessageStore { public MessageExt lookMessageByOffset(long commitLogOffset) { SelectMappedBufferResult sbr = this.commitLog.getMessage(commitLogOffset, 4); if (null != sbr) { try { int size = sbr.getByteBuffer().getInt(); return lookMessageByOffset(commitLogOffset, size); } finally { sbr.release(); } } return null; } DefaultMessageStore(final MessageStoreConfig messageStoreConfig, final BrokerStatsManager brokerStatsManager,
        final MessageArrivingListener messageArrivingListener, final BrokerConfig brokerConfig); }
DefaultMessageStore implements MessageStore { public MessageExt lookMessageByOffset(long commitLogOffset) { SelectMappedBufferResult sbr = this.commitLog.getMessage(commitLogOffset, 4); if (null != sbr) { try { int size = sbr.getByteBuffer().getInt(); return lookMessageByOffset(commitLogOffset, size); } finally { sbr.release(); } } return null; } 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); }
DefaultMessageStore implements MessageStore { public MessageExt lookMessageByOffset(long commitLogOffset) { SelectMappedBufferResult sbr = this.commitLog.getMessage(commitLogOffset, 4); if (null != sbr) { try { int size = sbr.getByteBuffer().getInt(); return lookMessageByOffset(commitLogOffset, size); } finally { sbr.release(); } } return null; } 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);  }