modify the map to ConcurrentHashMap

cat-consumer/src/main/java/com/dianping/cat/consumer/RealtimeConsumer.java

@@ -346,6 +346,8 @@ public class RealtimeConsumer extends ContainerHolder implements MessageConsumer
 							startPeriod(value);
 						} else {
 							endPeriod(-value);
+							// last period is over,make it asynchronous
+							Threads.forGroup("Cat").start(new EndTaskThread(-value));
 						}
 					} catch (Throwable e) {
 						Cat.logError(e);
@@ -374,5 +376,28 @@ public class RealtimeConsumer extends ContainerHolder implements MessageConsumer
 			m_periods.add(period);
 			period.start();
 		}
+
+		private class EndTaskThread implements Task {
+
+			private long m_startTime;
+
+			public EndTaskThread(long startTime) {
+				m_startTime = startTime;
+			}
+
+			@Override
+			public void run() {
+				endPeriod(m_startTime);
+			}
+
+			@Override
+			public String getName() {
+				return "End-Consumer-Task";
+			}
+
+			@Override
+			public void shutdown() {
+			}
+		}
 	}
 }
\ No newline at end of file

cat-consumer/src/main/java/com/dianping/cat/consumer/state/StateAnalyzer.java

@@ -45,18 +45,15 @@ public class StateAnalyzer extends AbstractMessageAnalyzer<StateReport> implemen
 		long minute = 1000 * 60;
 		long start = m_startTime;
 		long end = m_startTime + minute * 60;
+		int size = 0;
+		double maxTps = 0;
 
 		if (end > System.currentTimeMillis()) {
 			end = System.currentTimeMillis();
 		}
-
-		int size = 0;
-		double maxTps = 0;
-
 		for (; start < end; start += minute) {
 			Statistic state = m_serverStateManager.findState(start);
 			Message temp = machine.findOrCreateMessage(start);
-
 			Map<String, AtomicLong> totals = state.getMessageTotals();
 			long messageTotal = state.getMessageTotal();
 			temp.setTotal(messageTotal);
@@ -78,6 +75,7 @@ public class StateAnalyzer extends AbstractMessageAnalyzer<StateReport> implemen
 				long value = entry.getValue().get();
 				ProcessDomain domain = machine.findOrCreateProcessDomain(key);
 				Detail detail = domain.findOrCreateDetail(start);
+				
 				if (totals.containsKey(key)) {
 					domain.setTotal(value + domain.getTotal());
 					detail.setTotal(value);

cat-core/src/main/java/com/dianping/cat/statistic/ServerStatistic.java

 package com.dianping.cat.statistic;
 
-import java.util.HashMap;
-import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicLong;
 
+import org.jboss.netty.util.internal.ConcurrentHashMap;
+
 public class ServerStatistic {
-	private Map<Long, Statistic> m_statistics = new LinkedHashMap<Long, Statistic>(100);
+	private Map<Long, Statistic> m_statistics = new ConcurrentHashMap<Long, Statistic>(100);
 
 	public Statistic findOrCreate(Long time) {
 		Statistic state = m_statistics.get(time);
@@ -34,11 +34,11 @@ public class ServerStatistic {
 
 		private long m_messageDumpLoss;
 
-		private Map<String, AtomicLong> m_messageTotals = new HashMap<String, AtomicLong>(256);
+		private Map<String, AtomicLong> m_messageTotals = new ConcurrentHashMap<String, AtomicLong>(256);
 
-		private Map<String, AtomicLong> m_messageTotalLosses = new HashMap<String, AtomicLong>(256);
+		private Map<String, AtomicLong> m_messageTotalLosses = new ConcurrentHashMap<String, AtomicLong>(256);
 
-		private Map<String, Double> m_messageSizes = new HashMap<String, Double>(256);
+		private Map<String, Double> m_messageSizes = new ConcurrentHashMap<String, Double>(256);
 
 		private double m_processDelaySum;
 
@@ -85,7 +85,7 @@ public class ServerStatistic {
 		public void addMessageTotal(String domain, long messageTotal) {
 			AtomicLong value = m_messageTotals.get(domain);
 			if (value != null) {
-				value.set(value.get()+messageTotal);
+				value.set(value.get() + messageTotal);
 			} else {
 				m_messageTotals.put(domain, new AtomicLong(messageTotal));
 			}
@@ -94,9 +94,9 @@ public class ServerStatistic {
 		public void addMessageTotalLoss(String domain, long messageTotalLoss) {
 			AtomicLong value = m_messageTotalLosses.get(domain);
 			if (value != null) {
-				value.set(value.get()+messageTotalLoss);
+				value.set(value.get() + messageTotalLoss);
 			} else {
-				m_messageTotalLosses.put(domain,  new AtomicLong(messageTotalLoss));
+				m_messageTotalLosses.put(domain, new AtomicLong(messageTotalLoss));
 			}
 		}
 
@@ -220,7 +220,6 @@ public class ServerStatistic {
 		public double getMessageSize() {
 			return m_messageSize;
 		}
-
 	}
 
 }