Throttled { public void trigger(Key key, Runnable r) { final Runnable wrappedRunnable = () -> { try { r.run(); } finally { final Instant after = Instant.now(); synchronized (actionMap) { actionMap.put(key, Action.toBeExecutedSomeTimeLater(after)); } } }; final Instant now = Instant.now(); synchronized (actionMap) { final Action action = actionMap.get(key); if (action != null) { if (!action.alreadyScheduled) { final Instant lastTime = action.getExecutionTime(); final Duration between = Duration.between(lastTime, now); if (between.getSeconds() < minIntervalInSeconds) { final long delay = minIntervalInSeconds - between.getSeconds(); actionMap.put(key, action.scheduled()); scheduledExecutor.schedule(wrappedRunnable, delay, TimeUnit.SECONDS); } else { actionMap.put(key, action.scheduled()); scheduledExecutor.execute(wrappedRunnable); } } } else { actionMap.put(key, Action.toBeExecutedASAP(now)); scheduledExecutor.execute(wrappedRunnable); } } } Throttled(long minIntervalInSeconds); void trigger(Key key, Runnable r);  }
@Test public void testTriggerDoesntTriggerTooOften() throws InterruptedException { counter.set(0); final Throttled<String> throttled = new Throttled<>(10); throttled.trigger("x", slowIncrement); assertEquals(0, counter.get()); throttled.trigger("x", slowIncrement); assertEquals(0, counter.get()); waitALittleToAllowExecutorToProcessRunnables(); assertEquals(1, counter.get()); for (int i = 0; i < 10; i++) { throttled.trigger("x", counter::incrementAndGet); } Thread.sleep(1100); assertEquals(1, counter.get()); }
@Test public void testTriggerDoesTriggerAfterInterval() throws InterruptedException { counter.set(0); final Throttled<String> throttled = new Throttled<>(1); throttled.trigger("x", slowIncrement); waitALittleToAllowExecutorToProcessRunnables(); assertEquals(1, counter.get()); Thread.sleep(2000); throttled.trigger("x", slowIncrement); waitALittleToAllowExecutorToProcessRunnables(); assertEquals(2, counter.get()); }