@Test public void testFillProcessQueueInfo() { ProcessQueue pq = new ProcessQueue(); pq.putMessage(createMessageList(102400)); ProcessQueueInfo processQueueInfo = new ProcessQueueInfo(); pq.fillProcessQueueInfo(processQueueInfo); assertThat(processQueueInfo.getCachedMsgSizeInMiB()).isEqualTo(12); pq.takeMessages(10000); pq.commit(); pq.fillProcessQueueInfo(processQueueInfo); assertThat(processQueueInfo.getCachedMsgSizeInMiB()).isEqualTo(10); pq.takeMessages(10000); pq.commit(); pq.fillProcessQueueInfo(processQueueInfo); assertThat(processQueueInfo.getCachedMsgSizeInMiB()).isEqualTo(9); pq.takeMessages(80000); pq.commit(); pq.fillProcessQueueInfo(processQueueInfo); assertThat(processQueueInfo.getCachedMsgSizeInMiB()).isEqualTo(0); }
public void fillProcessQueueInfo(final ProcessQueueInfo info) { try { this.lockTreeMap.readLock().lockInterruptibly(); if (!this.msgTreeMap.isEmpty()) { info.setCachedMsgMinOffset(this.msgTreeMap.firstKey()); info.setCachedMsgMaxOffset(this.msgTreeMap.lastKey()); info.setCachedMsgCount(this.msgTreeMap.size()); info.setCachedMsgSizeInMiB((int) (this.msgSize.get() / (1024 * 1024))); } if (!this.consumingMsgOrderlyTreeMap.isEmpty()) { info.setTransactionMsgMinOffset(this.consumingMsgOrderlyTreeMap.firstKey()); info.setTransactionMsgMaxOffset(this.consumingMsgOrderlyTreeMap.lastKey()); info.setTransactionMsgCount(this.consumingMsgOrderlyTreeMap.size()); } info.setLocked(this.locked); info.setTryUnlockTimes(this.tryUnlockTimes.get()); info.setLastLockTimestamp(this.lastLockTimestamp); info.setDroped(this.dropped); info.setLastPullTimestamp(this.lastPullTimestamp); info.setLastConsumeTimestamp(this.lastConsumeTimestamp); } catch (Exception e) { } finally { this.lockTreeMap.readLock().unlock(); } }
ProcessQueue { public void fillProcessQueueInfo(final ProcessQueueInfo info) { try { this.lockTreeMap.readLock().lockInterruptibly(); if (!this.msgTreeMap.isEmpty()) { info.setCachedMsgMinOffset(this.msgTreeMap.firstKey()); info.setCachedMsgMaxOffset(this.msgTreeMap.lastKey()); info.setCachedMsgCount(this.msgTreeMap.size()); info.setCachedMsgSizeInMiB((int) (this.msgSize.get() / (1024 * 1024))); } if (!this.consumingMsgOrderlyTreeMap.isEmpty()) { info.setTransactionMsgMinOffset(this.consumingMsgOrderlyTreeMap.firstKey()); info.setTransactionMsgMaxOffset(this.consumingMsgOrderlyTreeMap.lastKey()); info.setTransactionMsgCount(this.consumingMsgOrderlyTreeMap.size()); } info.setLocked(this.locked); info.setTryUnlockTimes(this.tryUnlockTimes.get()); info.setLastLockTimestamp(this.lastLockTimestamp); info.setDroped(this.dropped); info.setLastPullTimestamp(this.lastPullTimestamp); info.setLastConsumeTimestamp(this.lastConsumeTimestamp); } catch (Exception e) { } finally { this.lockTreeMap.readLock().unlock(); } } }
ProcessQueue { public void fillProcessQueueInfo(final ProcessQueueInfo info) { try { this.lockTreeMap.readLock().lockInterruptibly(); if (!this.msgTreeMap.isEmpty()) { info.setCachedMsgMinOffset(this.msgTreeMap.firstKey()); info.setCachedMsgMaxOffset(this.msgTreeMap.lastKey()); info.setCachedMsgCount(this.msgTreeMap.size()); info.setCachedMsgSizeInMiB((int) (this.msgSize.get() / (1024 * 1024))); } if (!this.consumingMsgOrderlyTreeMap.isEmpty()) { info.setTransactionMsgMinOffset(this.consumingMsgOrderlyTreeMap.firstKey()); info.setTransactionMsgMaxOffset(this.consumingMsgOrderlyTreeMap.lastKey()); info.setTransactionMsgCount(this.consumingMsgOrderlyTreeMap.size()); } info.setLocked(this.locked); info.setTryUnlockTimes(this.tryUnlockTimes.get()); info.setLastLockTimestamp(this.lastLockTimestamp); info.setDroped(this.dropped); info.setLastPullTimestamp(this.lastPullTimestamp); info.setLastConsumeTimestamp(this.lastConsumeTimestamp); } catch (Exception e) { } finally { this.lockTreeMap.readLock().unlock(); } }  }
ProcessQueue { public void fillProcessQueueInfo(final ProcessQueueInfo info) { try { this.lockTreeMap.readLock().lockInterruptibly(); if (!this.msgTreeMap.isEmpty()) { info.setCachedMsgMinOffset(this.msgTreeMap.firstKey()); info.setCachedMsgMaxOffset(this.msgTreeMap.lastKey()); info.setCachedMsgCount(this.msgTreeMap.size()); info.setCachedMsgSizeInMiB((int) (this.msgSize.get() / (1024 * 1024))); } if (!this.consumingMsgOrderlyTreeMap.isEmpty()) { info.setTransactionMsgMinOffset(this.consumingMsgOrderlyTreeMap.firstKey()); info.setTransactionMsgMaxOffset(this.consumingMsgOrderlyTreeMap.lastKey()); info.setTransactionMsgCount(this.consumingMsgOrderlyTreeMap.size()); } info.setLocked(this.locked); info.setTryUnlockTimes(this.tryUnlockTimes.get()); info.setLastLockTimestamp(this.lastLockTimestamp); info.setDroped(this.dropped); info.setLastPullTimestamp(this.lastPullTimestamp); info.setLastConsumeTimestamp(this.lastConsumeTimestamp); } catch (Exception e) { } finally { this.lockTreeMap.readLock().unlock(); } }  boolean isLockExpired(); boolean isPullExpired(); void cleanExpiredMsg(DefaultMQPushConsumer pushConsumer); boolean putMessage(final List<MessageExt> msgs); long getMaxSpan(); long removeMessage(final List<MessageExt> msgs); TreeMap<Long, MessageExt> getMsgTreeMap(); AtomicLong getMsgCount(); AtomicLong getMsgSize(); boolean isDropped(); void setDropped(boolean dropped); boolean isLocked(); void setLocked(boolean locked); void rollback(); long commit(); void makeMessageToCosumeAgain(List<MessageExt> msgs); List<MessageExt> takeMessages(final int batchSize); boolean hasTempMessage(); void clear(); long getLastLockTimestamp(); void setLastLockTimestamp(long lastLockTimestamp); Lock getLockConsume(); long getLastPullTimestamp(); void setLastPullTimestamp(long lastPullTimestamp); long getMsgAccCnt(); void setMsgAccCnt(long msgAccCnt); long getTryUnlockTimes(); void incTryUnlockTimes(); void fillProcessQueueInfo(final ProcessQueueInfo info); long getLastConsumeTimestamp(); void setLastConsumeTimestamp(long lastConsumeTimestamp); }
ProcessQueue { public void fillProcessQueueInfo(final ProcessQueueInfo info) { try { this.lockTreeMap.readLock().lockInterruptibly(); if (!this.msgTreeMap.isEmpty()) { info.setCachedMsgMinOffset(this.msgTreeMap.firstKey()); info.setCachedMsgMaxOffset(this.msgTreeMap.lastKey()); info.setCachedMsgCount(this.msgTreeMap.size()); info.setCachedMsgSizeInMiB((int) (this.msgSize.get() / (1024 * 1024))); } if (!this.consumingMsgOrderlyTreeMap.isEmpty()) { info.setTransactionMsgMinOffset(this.consumingMsgOrderlyTreeMap.firstKey()); info.setTransactionMsgMaxOffset(this.consumingMsgOrderlyTreeMap.lastKey()); info.setTransactionMsgCount(this.consumingMsgOrderlyTreeMap.size()); } info.setLocked(this.locked); info.setTryUnlockTimes(this.tryUnlockTimes.get()); info.setLastLockTimestamp(this.lastLockTimestamp); info.setDroped(this.dropped); info.setLastPullTimestamp(this.lastPullTimestamp); info.setLastConsumeTimestamp(this.lastConsumeTimestamp); } catch (Exception e) { } finally { this.lockTreeMap.readLock().unlock(); } }  boolean isLockExpired(); boolean isPullExpired(); void cleanExpiredMsg(DefaultMQPushConsumer pushConsumer); boolean putMessage(final List<MessageExt> msgs); long getMaxSpan(); long removeMessage(final List<MessageExt> msgs); TreeMap<Long, MessageExt> getMsgTreeMap(); AtomicLong getMsgCount(); AtomicLong getMsgSize(); boolean isDropped(); void setDropped(boolean dropped); boolean isLocked(); void setLocked(boolean locked); void rollback(); long commit(); void makeMessageToCosumeAgain(List<MessageExt> msgs); List<MessageExt> takeMessages(final int batchSize); boolean hasTempMessage(); void clear(); long getLastLockTimestamp(); void setLastLockTimestamp(long lastLockTimestamp); Lock getLockConsume(); long getLastPullTimestamp(); void setLastPullTimestamp(long lastPullTimestamp); long getMsgAccCnt(); void setMsgAccCnt(long msgAccCnt); long getTryUnlockTimes(); void incTryUnlockTimes(); void fillProcessQueueInfo(final ProcessQueueInfo info); long getLastConsumeTimestamp(); void setLastConsumeTimestamp(long lastConsumeTimestamp); final static long REBALANCE_LOCK_MAX_LIVE_TIME; final static long REBALANCE_LOCK_INTERVAL; }