KeyedOneThreadTaskExecutor implements Destroyable { public void execute(K key, Command<?> command){ OneThreadTaskExecutor oneThreadTaskExecutor = getOrCreate(key); oneThreadTaskExecutor.executeCommand(command); } KeyedOneThreadTaskExecutor(Executor executors); void execute(K key, Command<?> command); @Override void destroy();  }
@Test public void testHang() throws TimeoutException, IOException { int threadCount = 10; int taskCount = threadCount; ExecutorService executorService = null; try{ executorService = Executors.newFixedThreadPool(threadCount, XpipeThreadFactory.create("test-hang")); keyed = new KeyedOneThreadTaskExecutor<>(executorService); AtomicInteger completeCount = new AtomicInteger(); for(int i=0; i<taskCount ;i++){ int finalI = i; ParallelCommandChain parallelCommandChain = new ParallelCommandChain(executorService); parallelCommandChain.add(new TestCommand("success:" + i, sleepInterval)); parallelCommandChain.future().addListener(new CommandFutureListener<Object>() { @Override public void operationComplete(CommandFuture<Object> commandFuture) throws Exception { logger.info("[operationComplete]{}", finalI); completeCount.incrementAndGet(); } }); keyed.execute(String.valueOf(i), parallelCommandChain); } waitConditionUntilTimeOut(() -> completeCount.get() == taskCount); }finally { executorService.shutdownNow(); } }
@Test public void testSameKey(){ TestCommand command1 = new TestCommand("success", sleepInterval); TestCommand command2 = new TestCommand("success", sleepInterval); keyed.execute("key1", command1); keyed.execute("key1", command2); sleep(sleepInterval/2); Assert.assertTrue(command1.isBeginExecute()); Assert.assertFalse(command2.isBeginExecute()); }
@Test public void testDifferentKey(){ TestCommand command1 = new TestCommand("success", sleepInterval); TestCommand command2 = new TestCommand("success", sleepInterval); keyed.execute("key1", command1); keyed.execute("key2", command2); sleep(sleepInterval/2); Assert.assertTrue(command1.isBeginExecute()); Assert.assertTrue(command2.isBeginExecute()); }