DistributedOperationQueue { public synchronized void addByThreadId(DistributedOperation distributedOperation, long threadId) { if (!allowAdd) return; ArrayList<DistributedOperation> operationListByKey = operationMapByKey.get(distributedOperation.getKey()); if (operationListByKey == null) { operationListByKey = new ArrayList<DistributedOperation>(); operationMapByKey.put(distributedOperation.getKey(), operationListByKey); } operationListByKey.add(distributedOperation); ArrayList<DistributedOperation> operationListByThreadId = operationByThreadId.get(threadId); if (operationListByThreadId == null) { operationListByThreadId = new ArrayList<DistributedOperation>(); operationByThreadId.put(threadId, operationListByThreadId); } operationListByThreadId.add(distributedOperation); }  boolean setAllowAdd(boolean allowAdd); synchronized DistributedOperation[] lockAndGetOperations(long threadId); synchronized void deleteAndUnlockOperations(DistributedOperation[] operations); synchronized void clear(); synchronized void add(DistributedOperation distributedOperation); synchronized void addByThreadId(DistributedOperation distributedOperation, long threadId); static final Logger logger; }
@Test public void testOperationQueue() throws InterruptedException{ Thread t0 = new Getter(queue), t1 =new Getter(queue), t2=new Getter(queue); long tid0 = t0.getId(),tid1 = t1.getId(),tid2=t2.getId(); synchronized(queue){ queue.addByThreadId(generateOprationById(0), tid0); queue.addByThreadId(generateOprationById(3), tid1); queue.addByThreadId(generateOprationById(2), tid2); queue.addByThreadId(generateOprationById(1), tid0); queue.addByThreadId(generateOprationById(1), tid1); queue.addByThreadId(generateOprationById(1), tid2); queue.addByThreadId(generateOprationById(2), tid1); queue.addByThreadId(generateOprationById(3), tid2); } t0.setName("t0"); t1.setName("t1"); t2.setName("t2"); t0.start(); Thread.sleep(5000); t2.start(); Thread.sleep(10000); t1.start(); t0.join(); t1.join(); t2.join(); }
@Test public void testResourceCycle() throws InterruptedException{ CountDownLatch fire = new CountDownLatch(1); CountDownLatch notifier = new CountDownLatch(1); CountDownLatch interrupter = new CountDownLatch(1); Thread t0 = new NotifyGetter(fire, notifier, interrupter, queue), t1 =new NotifyGetter(fire, notifier, interrupter, queue), t2=new NotifyGetter(fire, notifier, interrupter, queue); long tid0 = t0.getId(),tid1 = t1.getId(),tid2=t2.getId(); synchronized(queue){ queue.addByThreadId(generateOprationById(0), tid0); queue.addByThreadId(generateOprationById(3), tid1); queue.addByThreadId(generateOprationById(2), tid2); queue.addByThreadId(generateOprationById(1), tid0); queue.addByThreadId(generateOprationById(1), tid1); queue.addByThreadId(generateOprationById(1), tid2); queue.addByThreadId(generateOprationById(2), tid1); queue.addByThreadId(generateOprationById(3), tid2); } t0.setName("t0"); t1.setName("t1"); t2.setName("t2"); t0.start(); t1.start(); t2.start(); Thread.sleep(3000); interrupter.countDown(); Thread.sleep(3000); fire.countDown(); t0.join(); t1.join(); t2.join(); }