@Test public void testListResourcesOfApp() { testSave(); System.out.println( "[" + System.currentTimeMillis() + "] Basic test data ready in testListResourcesOfApp"); List<CompletableFuture> futures = Lists.newArrayList(); final CyclicBarrier cyclicBarrier = new CyclicBarrier(8); for (int j = 0; j < 10000; j++) { futures.add( CompletableFuture.runAsync(() -> { try { cyclicBarrier.await(); inMemoryMetricsRepository.listResourcesOfApp(DEFAULT_APP); } catch (InterruptedException | BrokenBarrierException e) { e.printStackTrace(); } }, executorService) ); } for (int i = 0; i < 10000; i++) { MetricEntity entry = new MetricEntity(); entry.setApp(DEFAULT_APP); entry.setResource(DEFAULT_RESOURCE); entry.setTimestamp(new Date(System.currentTimeMillis() - EXPIRE_TIME - 1000L)); entry.setPassQps(1L); entry.setExceptionQps(1L); entry.setBlockQps(0L); entry.setSuccessQps(1L); inMemoryMetricsRepository.save(entry); } CompletableFuture all = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])); try { all.get(10, TimeUnit.SECONDS); } catch (InterruptedException e) { e.printStackTrace(); } catch (ExecutionException e) { e.getCause().printStackTrace(); if (e.getCause() instanceof ConcurrentModificationException) { fail("concurrent error occurred"); } else { fail("unexpected exception"); } } catch (TimeoutException e) { fail("allOf future timeout"); } }
@Override public List<String> listResourcesOfApp(String app) { List<String> results = new ArrayList<>(); if (StringUtil.isBlank(app)) { return results; } Map<String, ConcurrentLinkedHashMap<Long, MetricEntity>> resourceMap = allMetrics.get(app); if (resourceMap == null) { return results; } final long minTimeMs = System.currentTimeMillis() - 1000 * 60; Map<String, MetricEntity> resourceCount = new ConcurrentHashMap<>(32); for (Entry<String, ConcurrentLinkedHashMap<Long, MetricEntity>> resourceMetrics : resourceMap.entrySet()) { for (Entry<Long, MetricEntity> metrics : resourceMetrics.getValue().entrySet()) { if (metrics.getKey() < minTimeMs) { continue; } MetricEntity newEntity = metrics.getValue(); if (resourceCount.containsKey(resourceMetrics.getKey())) { MetricEntity oldEntity = resourceCount.get(resourceMetrics.getKey()); oldEntity.addPassQps(newEntity.getPassQps()); oldEntity.addRtAndSuccessQps(newEntity.getRt(), newEntity.getSuccessQps()); oldEntity.addBlockQps(newEntity.getBlockQps()); oldEntity.addExceptionQps(newEntity.getExceptionQps()); oldEntity.addCount(1); } else { resourceCount.put(resourceMetrics.getKey(), MetricEntity.copyOf(newEntity)); } } } return resourceCount.entrySet() .stream() .sorted((o1, o2) -> { MetricEntity e1 = o1.getValue(); MetricEntity e2 = o2.getValue(); int t = e2.getBlockQps().compareTo(e1.getBlockQps()); if (t != 0) { return t; } return e2.getPassQps().compareTo(e1.getPassQps()); }) .map(Entry::getKey) .collect(Collectors.toList()); }
InMemoryMetricsRepository implements MetricsRepository<MetricEntity> { @Override public List<String> listResourcesOfApp(String app) { List<String> results = new ArrayList<>(); if (StringUtil.isBlank(app)) { return results; } Map<String, ConcurrentLinkedHashMap<Long, MetricEntity>> resourceMap = allMetrics.get(app); if (resourceMap == null) { return results; } final long minTimeMs = System.currentTimeMillis() - 1000 * 60; Map<String, MetricEntity> resourceCount = new ConcurrentHashMap<>(32); for (Entry<String, ConcurrentLinkedHashMap<Long, MetricEntity>> resourceMetrics : resourceMap.entrySet()) { for (Entry<Long, MetricEntity> metrics : resourceMetrics.getValue().entrySet()) { if (metrics.getKey() < minTimeMs) { continue; } MetricEntity newEntity = metrics.getValue(); if (resourceCount.containsKey(resourceMetrics.getKey())) { MetricEntity oldEntity = resourceCount.get(resourceMetrics.getKey()); oldEntity.addPassQps(newEntity.getPassQps()); oldEntity.addRtAndSuccessQps(newEntity.getRt(), newEntity.getSuccessQps()); oldEntity.addBlockQps(newEntity.getBlockQps()); oldEntity.addExceptionQps(newEntity.getExceptionQps()); oldEntity.addCount(1); } else { resourceCount.put(resourceMetrics.getKey(), MetricEntity.copyOf(newEntity)); } } } return resourceCount.entrySet() .stream() .sorted((o1, o2) -> { MetricEntity e1 = o1.getValue(); MetricEntity e2 = o2.getValue(); int t = e2.getBlockQps().compareTo(e1.getBlockQps()); if (t != 0) { return t; } return e2.getPassQps().compareTo(e1.getPassQps()); }) .map(Entry::getKey) .collect(Collectors.toList()); } }
InMemoryMetricsRepository implements MetricsRepository<MetricEntity> { @Override public List<String> listResourcesOfApp(String app) { List<String> results = new ArrayList<>(); if (StringUtil.isBlank(app)) { return results; } Map<String, ConcurrentLinkedHashMap<Long, MetricEntity>> resourceMap = allMetrics.get(app); if (resourceMap == null) { return results; } final long minTimeMs = System.currentTimeMillis() - 1000 * 60; Map<String, MetricEntity> resourceCount = new ConcurrentHashMap<>(32); for (Entry<String, ConcurrentLinkedHashMap<Long, MetricEntity>> resourceMetrics : resourceMap.entrySet()) { for (Entry<Long, MetricEntity> metrics : resourceMetrics.getValue().entrySet()) { if (metrics.getKey() < minTimeMs) { continue; } MetricEntity newEntity = metrics.getValue(); if (resourceCount.containsKey(resourceMetrics.getKey())) { MetricEntity oldEntity = resourceCount.get(resourceMetrics.getKey()); oldEntity.addPassQps(newEntity.getPassQps()); oldEntity.addRtAndSuccessQps(newEntity.getRt(), newEntity.getSuccessQps()); oldEntity.addBlockQps(newEntity.getBlockQps()); oldEntity.addExceptionQps(newEntity.getExceptionQps()); oldEntity.addCount(1); } else { resourceCount.put(resourceMetrics.getKey(), MetricEntity.copyOf(newEntity)); } } } return resourceCount.entrySet() .stream() .sorted((o1, o2) -> { MetricEntity e1 = o1.getValue(); MetricEntity e2 = o2.getValue(); int t = e2.getBlockQps().compareTo(e1.getBlockQps()); if (t != 0) { return t; } return e2.getPassQps().compareTo(e1.getPassQps()); }) .map(Entry::getKey) .collect(Collectors.toList()); }  }
InMemoryMetricsRepository implements MetricsRepository<MetricEntity> { @Override public List<String> listResourcesOfApp(String app) { List<String> results = new ArrayList<>(); if (StringUtil.isBlank(app)) { return results; } Map<String, ConcurrentLinkedHashMap<Long, MetricEntity>> resourceMap = allMetrics.get(app); if (resourceMap == null) { return results; } final long minTimeMs = System.currentTimeMillis() - 1000 * 60; Map<String, MetricEntity> resourceCount = new ConcurrentHashMap<>(32); for (Entry<String, ConcurrentLinkedHashMap<Long, MetricEntity>> resourceMetrics : resourceMap.entrySet()) { for (Entry<Long, MetricEntity> metrics : resourceMetrics.getValue().entrySet()) { if (metrics.getKey() < minTimeMs) { continue; } MetricEntity newEntity = metrics.getValue(); if (resourceCount.containsKey(resourceMetrics.getKey())) { MetricEntity oldEntity = resourceCount.get(resourceMetrics.getKey()); oldEntity.addPassQps(newEntity.getPassQps()); oldEntity.addRtAndSuccessQps(newEntity.getRt(), newEntity.getSuccessQps()); oldEntity.addBlockQps(newEntity.getBlockQps()); oldEntity.addExceptionQps(newEntity.getExceptionQps()); oldEntity.addCount(1); } else { resourceCount.put(resourceMetrics.getKey(), MetricEntity.copyOf(newEntity)); } } } return resourceCount.entrySet() .stream() .sorted((o1, o2) -> { MetricEntity e1 = o1.getValue(); MetricEntity e2 = o2.getValue(); int t = e2.getBlockQps().compareTo(e1.getBlockQps()); if (t != 0) { return t; } return e2.getPassQps().compareTo(e1.getPassQps()); }) .map(Entry::getKey) .collect(Collectors.toList()); }  @Override synchronized void save(MetricEntity entity); @Override synchronized void saveAll(Iterable<MetricEntity> metrics); @Override synchronized List<MetricEntity> queryByAppAndResourceBetween(String app, String resource,
                                                                        long startTime, long endTime); @Override List<String> listResourcesOfApp(String app); }
InMemoryMetricsRepository implements MetricsRepository<MetricEntity> { @Override public List<String> listResourcesOfApp(String app) { List<String> results = new ArrayList<>(); if (StringUtil.isBlank(app)) { return results; } Map<String, ConcurrentLinkedHashMap<Long, MetricEntity>> resourceMap = allMetrics.get(app); if (resourceMap == null) { return results; } final long minTimeMs = System.currentTimeMillis() - 1000 * 60; Map<String, MetricEntity> resourceCount = new ConcurrentHashMap<>(32); for (Entry<String, ConcurrentLinkedHashMap<Long, MetricEntity>> resourceMetrics : resourceMap.entrySet()) { for (Entry<Long, MetricEntity> metrics : resourceMetrics.getValue().entrySet()) { if (metrics.getKey() < minTimeMs) { continue; } MetricEntity newEntity = metrics.getValue(); if (resourceCount.containsKey(resourceMetrics.getKey())) { MetricEntity oldEntity = resourceCount.get(resourceMetrics.getKey()); oldEntity.addPassQps(newEntity.getPassQps()); oldEntity.addRtAndSuccessQps(newEntity.getRt(), newEntity.getSuccessQps()); oldEntity.addBlockQps(newEntity.getBlockQps()); oldEntity.addExceptionQps(newEntity.getExceptionQps()); oldEntity.addCount(1); } else { resourceCount.put(resourceMetrics.getKey(), MetricEntity.copyOf(newEntity)); } } } return resourceCount.entrySet() .stream() .sorted((o1, o2) -> { MetricEntity e1 = o1.getValue(); MetricEntity e2 = o2.getValue(); int t = e2.getBlockQps().compareTo(e1.getBlockQps()); if (t != 0) { return t; } return e2.getPassQps().compareTo(e1.getPassQps()); }) .map(Entry::getKey) .collect(Collectors.toList()); }  @Override synchronized void save(MetricEntity entity); @Override synchronized void saveAll(Iterable<MetricEntity> metrics); @Override synchronized List<MetricEntity> queryByAppAndResourceBetween(String app, String resource,
                                                                        long startTime, long endTime); @Override List<String> listResourcesOfApp(String app);  }