DefaultMessageStore implements MessageStore { public PutMessageResult putMessage(MessageExtBrokerInner msg) { if (this.shutdown) { log.warn("message store has shutdown, so putMessage is forbidden"); return new PutMessageResult(PutMessageStatus.SERVICE_NOT_AVAILABLE, null); } if (BrokerRole.SLAVE == this.messageStoreConfig.getBrokerRole()) { long value = this.printTimes.getAndIncrement(); if ((value % 50000) == 0) { log.warn("message store is slave mode, so putMessage is forbidden "); } return new PutMessageResult(PutMessageStatus.SERVICE_NOT_AVAILABLE, null); } if (!this.runningFlags.isWriteable()) { long value = this.printTimes.getAndIncrement(); if ((value % 50000) == 0) { log.warn("message store is not writeable, so putMessage is forbidden " + this.runningFlags.getFlagBits()); } return new PutMessageResult(PutMessageStatus.SERVICE_NOT_AVAILABLE, null); } else { this.printTimes.set(0); } if (msg.getTopic().length() > Byte.MAX_VALUE) { log.warn("putMessage message topic length too long " + msg.getTopic().length()); return new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, null); } if (msg.getPropertiesString() != null && msg.getPropertiesString().length() > Short.MAX_VALUE) { log.warn("putMessage message properties length too long " + msg.getPropertiesString().length()); return new PutMessageResult(PutMessageStatus.PROPERTIES_SIZE_EXCEEDED, null); } if (this.isOSPageCacheBusy()) { return new PutMessageResult(PutMessageStatus.OS_PAGECACHE_BUSY, null); } long beginTime = this.getSystemClock().now(); PutMessageResult result = this.commitLog.putMessage(msg); long eclipseTime = this.getSystemClock().now() - beginTime; if (eclipseTime > 500) { log.warn("putMessage not in lock eclipse time(ms)={}, bodyLength={}", eclipseTime, msg.getBody().length); } this.storeStatsService.setPutMessageEntireTimeMax(eclipseTime); if (null == result || !result.isOk()) { this.storeStatsService.getPutMessageFailedTimes().incrementAndGet(); } return result; } 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 testWriteScheduleMsg() throws Exception { long totalMsgs = 60000; QUEUE_TOTAL = 10; MessageBody = StoreMessage.getBytes(); for (long i = 0; i < totalMsgs; i++) { MessageExtBrokerInner msgInner=buildScheduleMessage(i); messageStore.putMessage(msgInner); } System.err.println(System.currentTimeMillis()); Thread.sleep(200000); }