DataContainer { public synchronized Map<String, Map<String, Object>> pop() { Map<String, Map<String, Object>> operations = new ConcurrentHashMap<String, Map<String, Object>>(); try { while (!hasFinished()) { logger.debug("Operation data is consumed completely."); this.wait(); } } catch (InterruptedException e) { logger.warn("Wait push data failed: " + e.getMessage()); } String id = ""; Map<String, Object> object = null; Iterator<Entry<String, Map<String, Object>>> operationIterator = dataQueue.entrySet().iterator(); Entry<String, Map<String, Object>> operation = null; while (operationIterator.hasNext()) { operation = operationIterator.next(); if (operation != null) { id = operation.getKey(); if (isFinished(id)) { object = operation.getValue(); operations.put(id, object); operationIterator.remove(); } } } this.notify(); return operations; }  void push(String id, String key, Object t); synchronized void push(String id, String key, Object t, boolean force); synchronized Map<String, Map<String, Object>> pop(); void setMaxLength(int maxLength);  }
@Test public void testPop() { final DataContainer dataContainer = new DataContainer(); Runnable t = new Runnable() { @Override public void run() { while(true) { Map<String, Map<String, Object>> data = dataContainer.pop(); assertEquals(data.size(), 2); assertEquals(data.get("1").size(), 6); assertEquals(data.get("2").size(), 6); if (data.size() == 2) { break; } } }}; Thread tr = new Thread(t); tr.start(); dataContainer.push("1", "aa", "01"); dataContainer.push("1", "bb", "02"); dataContainer.push("1", "cc", "03"); dataContainer.push("1", "dd", "04"); dataContainer.push("1", "ee", "05"); dataContainer.push("1", "ff", "06"); dataContainer.push("2", "aa", "01"); dataContainer.push("2", "bb", "02"); dataContainer.push("2", "cc", "03"); dataContainer.push("2", "dd", "04"); dataContainer.push("2", "ee", "05"); dataContainer.push("2", "ff", "06"); try { tr.join(); } catch (InterruptedException e) { e.printStackTrace(); } }