@Test public void testCounter() throws InterruptedException { System.out.println("start time " + System.currentTimeMillis()); CountDownLatch latch = new CountDownLatch(1); final CountActor[] countActors = new CountActor[size]; SumActor sumActor = new SumActor(size, latch); for (int i = 0; i < size; i++) { countActors[i] = new CountActor(sumActor); final int k = i; new Thread(() -> { for (int j = 0; j < 1000; j++) { countActors[k].addMessage((long) j); } countActors[k].addMessage((long) -1); }).start(); } latch.await(); }
public void addMessage(E message) { queue.offer(message); GateKeeper.readyToExecute(this); }
Actor implements IActor { public void addMessage(E message) { queue.offer(message); GateKeeper.readyToExecute(this); } }
Actor implements IActor { public void addMessage(E message) { queue.offer(message); GateKeeper.readyToExecute(this); } Actor(); }
Actor implements IActor { public void addMessage(E message) { queue.offer(message); GateKeeper.readyToExecute(this); } Actor(); @Override long messageCount(); @Override boolean isEnded(); void stop(); @Override AtomicInteger getStatus(); abstract void doWork(E message); @Override void execute(); void addMessage(E message); }
Actor implements IActor { public void addMessage(E message) { queue.offer(message); GateKeeper.readyToExecute(this); } Actor(); @Override long messageCount(); @Override boolean isEnded(); void stop(); @Override AtomicInteger getStatus(); abstract void doWork(E message); @Override void execute(); void addMessage(E message);  }