@Test public void testAnalyzeProcessQueue(){ String result = ConsumerRunningInfo.analyzeProcessQueue("client_id", consumerRunningInfo); assertThat(result).isEmpty(); }
public static String analyzeProcessQueue(final String clientId, ConsumerRunningInfo info) { StringBuilder sb = new StringBuilder(); boolean push = false; { String property = info.getProperties().getProperty(ConsumerRunningInfo.PROP_CONSUME_TYPE); if (property == null) { property = ((ConsumeType) info.getProperties().get(ConsumerRunningInfo.PROP_CONSUME_TYPE)).name(); } push = ConsumeType.valueOf(property) == ConsumeType.CONSUME_PASSIVELY; } boolean orderMsg = false; { String property = info.getProperties().getProperty(ConsumerRunningInfo.PROP_CONSUME_ORDERLY); orderMsg = Boolean.parseBoolean(property); } if (push) { Iterator<Entry<MessageQueue, ProcessQueueInfo>> it = info.getMqTable().entrySet().iterator(); while (it.hasNext()) { Entry<MessageQueue, ProcessQueueInfo> next = it.next(); MessageQueue mq = next.getKey(); ProcessQueueInfo pq = next.getValue(); if (orderMsg) { if (!pq.isLocked()) { sb.append(String.format("%s %s can't lock for a while, %dms%n", clientId, mq, System.currentTimeMillis() - pq.getLastLockTimestamp())); } else { if (pq.isDroped() && (pq.getTryUnlockTimes() > 0)) { sb.append(String.format("%s %s unlock %d times, still failed%n", clientId, mq, pq.getTryUnlockTimes())); } } } else { long diff = System.currentTimeMillis() - pq.getLastConsumeTimestamp(); if (diff > (1000 * 60) && pq.getCachedMsgCount() > 0) { sb.append(String.format("%s %s can't consume for a while, maybe blocked, %dms%n", clientId, mq, diff)); } } } } return sb.toString(); }
ConsumerRunningInfo extends RemotingSerializable { public static String analyzeProcessQueue(final String clientId, ConsumerRunningInfo info) { StringBuilder sb = new StringBuilder(); boolean push = false; { String property = info.getProperties().getProperty(ConsumerRunningInfo.PROP_CONSUME_TYPE); if (property == null) { property = ((ConsumeType) info.getProperties().get(ConsumerRunningInfo.PROP_CONSUME_TYPE)).name(); } push = ConsumeType.valueOf(property) == ConsumeType.CONSUME_PASSIVELY; } boolean orderMsg = false; { String property = info.getProperties().getProperty(ConsumerRunningInfo.PROP_CONSUME_ORDERLY); orderMsg = Boolean.parseBoolean(property); } if (push) { Iterator<Entry<MessageQueue, ProcessQueueInfo>> it = info.getMqTable().entrySet().iterator(); while (it.hasNext()) { Entry<MessageQueue, ProcessQueueInfo> next = it.next(); MessageQueue mq = next.getKey(); ProcessQueueInfo pq = next.getValue(); if (orderMsg) { if (!pq.isLocked()) { sb.append(String.format("%s %s can't lock for a while, %dms%n", clientId, mq, System.currentTimeMillis() - pq.getLastLockTimestamp())); } else { if (pq.isDroped() && (pq.getTryUnlockTimes() > 0)) { sb.append(String.format("%s %s unlock %d times, still failed%n", clientId, mq, pq.getTryUnlockTimes())); } } } else { long diff = System.currentTimeMillis() - pq.getLastConsumeTimestamp(); if (diff > (1000 * 60) && pq.getCachedMsgCount() > 0) { sb.append(String.format("%s %s can't consume for a while, maybe blocked, %dms%n", clientId, mq, diff)); } } } } return sb.toString(); } }
ConsumerRunningInfo extends RemotingSerializable { public static String analyzeProcessQueue(final String clientId, ConsumerRunningInfo info) { StringBuilder sb = new StringBuilder(); boolean push = false; { String property = info.getProperties().getProperty(ConsumerRunningInfo.PROP_CONSUME_TYPE); if (property == null) { property = ((ConsumeType) info.getProperties().get(ConsumerRunningInfo.PROP_CONSUME_TYPE)).name(); } push = ConsumeType.valueOf(property) == ConsumeType.CONSUME_PASSIVELY; } boolean orderMsg = false; { String property = info.getProperties().getProperty(ConsumerRunningInfo.PROP_CONSUME_ORDERLY); orderMsg = Boolean.parseBoolean(property); } if (push) { Iterator<Entry<MessageQueue, ProcessQueueInfo>> it = info.getMqTable().entrySet().iterator(); while (it.hasNext()) { Entry<MessageQueue, ProcessQueueInfo> next = it.next(); MessageQueue mq = next.getKey(); ProcessQueueInfo pq = next.getValue(); if (orderMsg) { if (!pq.isLocked()) { sb.append(String.format("%s %s can't lock for a while, %dms%n", clientId, mq, System.currentTimeMillis() - pq.getLastLockTimestamp())); } else { if (pq.isDroped() && (pq.getTryUnlockTimes() > 0)) { sb.append(String.format("%s %s unlock %d times, still failed%n", clientId, mq, pq.getTryUnlockTimes())); } } } else { long diff = System.currentTimeMillis() - pq.getLastConsumeTimestamp(); if (diff > (1000 * 60) && pq.getCachedMsgCount() > 0) { sb.append(String.format("%s %s can't consume for a while, maybe blocked, %dms%n", clientId, mq, diff)); } } } } return sb.toString(); }  }
ConsumerRunningInfo extends RemotingSerializable { public static String analyzeProcessQueue(final String clientId, ConsumerRunningInfo info) { StringBuilder sb = new StringBuilder(); boolean push = false; { String property = info.getProperties().getProperty(ConsumerRunningInfo.PROP_CONSUME_TYPE); if (property == null) { property = ((ConsumeType) info.getProperties().get(ConsumerRunningInfo.PROP_CONSUME_TYPE)).name(); } push = ConsumeType.valueOf(property) == ConsumeType.CONSUME_PASSIVELY; } boolean orderMsg = false; { String property = info.getProperties().getProperty(ConsumerRunningInfo.PROP_CONSUME_ORDERLY); orderMsg = Boolean.parseBoolean(property); } if (push) { Iterator<Entry<MessageQueue, ProcessQueueInfo>> it = info.getMqTable().entrySet().iterator(); while (it.hasNext()) { Entry<MessageQueue, ProcessQueueInfo> next = it.next(); MessageQueue mq = next.getKey(); ProcessQueueInfo pq = next.getValue(); if (orderMsg) { if (!pq.isLocked()) { sb.append(String.format("%s %s can't lock for a while, %dms%n", clientId, mq, System.currentTimeMillis() - pq.getLastLockTimestamp())); } else { if (pq.isDroped() && (pq.getTryUnlockTimes() > 0)) { sb.append(String.format("%s %s unlock %d times, still failed%n", clientId, mq, pq.getTryUnlockTimes())); } } } else { long diff = System.currentTimeMillis() - pq.getLastConsumeTimestamp(); if (diff > (1000 * 60) && pq.getCachedMsgCount() > 0) { sb.append(String.format("%s %s can't consume for a while, maybe blocked, %dms%n", clientId, mq, diff)); } } } } return sb.toString(); }  static boolean analyzeSubscription(final TreeMap<String/* clientId */, ConsumerRunningInfo> criTable); static boolean analyzeRebalance(final TreeMap<String/* clientId */, ConsumerRunningInfo> criTable); static String analyzeProcessQueue(final String clientId, ConsumerRunningInfo info); Properties getProperties(); void setProperties(Properties properties); TreeSet<SubscriptionData> getSubscriptionSet(); void setSubscriptionSet(TreeSet<SubscriptionData> subscriptionSet); TreeMap<MessageQueue, ProcessQueueInfo> getMqTable(); void setMqTable(TreeMap<MessageQueue, ProcessQueueInfo> mqTable); TreeMap<String, ConsumeStatus> getStatusTable(); void setStatusTable(TreeMap<String, ConsumeStatus> statusTable); String formatString(); String getJstack(); void setJstack(String jstack); }
ConsumerRunningInfo extends RemotingSerializable { public static String analyzeProcessQueue(final String clientId, ConsumerRunningInfo info) { StringBuilder sb = new StringBuilder(); boolean push = false; { String property = info.getProperties().getProperty(ConsumerRunningInfo.PROP_CONSUME_TYPE); if (property == null) { property = ((ConsumeType) info.getProperties().get(ConsumerRunningInfo.PROP_CONSUME_TYPE)).name(); } push = ConsumeType.valueOf(property) == ConsumeType.CONSUME_PASSIVELY; } boolean orderMsg = false; { String property = info.getProperties().getProperty(ConsumerRunningInfo.PROP_CONSUME_ORDERLY); orderMsg = Boolean.parseBoolean(property); } if (push) { Iterator<Entry<MessageQueue, ProcessQueueInfo>> it = info.getMqTable().entrySet().iterator(); while (it.hasNext()) { Entry<MessageQueue, ProcessQueueInfo> next = it.next(); MessageQueue mq = next.getKey(); ProcessQueueInfo pq = next.getValue(); if (orderMsg) { if (!pq.isLocked()) { sb.append(String.format("%s %s can't lock for a while, %dms%n", clientId, mq, System.currentTimeMillis() - pq.getLastLockTimestamp())); } else { if (pq.isDroped() && (pq.getTryUnlockTimes() > 0)) { sb.append(String.format("%s %s unlock %d times, still failed%n", clientId, mq, pq.getTryUnlockTimes())); } } } else { long diff = System.currentTimeMillis() - pq.getLastConsumeTimestamp(); if (diff > (1000 * 60) && pq.getCachedMsgCount() > 0) { sb.append(String.format("%s %s can't consume for a while, maybe blocked, %dms%n", clientId, mq, diff)); } } } } return sb.toString(); }  static boolean analyzeSubscription(final TreeMap<String/* clientId */, ConsumerRunningInfo> criTable); static boolean analyzeRebalance(final TreeMap<String/* clientId */, ConsumerRunningInfo> criTable); static String analyzeProcessQueue(final String clientId, ConsumerRunningInfo info); Properties getProperties(); void setProperties(Properties properties); TreeSet<SubscriptionData> getSubscriptionSet(); void setSubscriptionSet(TreeSet<SubscriptionData> subscriptionSet); TreeMap<MessageQueue, ProcessQueueInfo> getMqTable(); void setMqTable(TreeMap<MessageQueue, ProcessQueueInfo> mqTable); TreeMap<String, ConsumeStatus> getStatusTable(); void setStatusTable(TreeMap<String, ConsumeStatus> statusTable); String formatString(); String getJstack(); void setJstack(String jstack); static final String PROP_NAMESERVER_ADDR; static final String PROP_THREADPOOL_CORE_SIZE; static final String PROP_CONSUME_ORDERLY; static final String PROP_CONSUME_TYPE; static final String PROP_CLIENT_VERSION; static final String PROP_CONSUMER_START_TIMESTAMP; }