Почему ReentrantLock не работает в SpringXD

Question

Я пытаюсь добавить ReentrantLock в мою работу над SpringXD, но иногда кажется, что есть перебои в потоках.

Например, у меня есть этот код:

public class LoadGenerator extends MessageProducerSupport { 

private final AtomicBoolean running = new AtomicBoolean(false); 
private ExecutorService executorService; 

Logger logger = LoggerFactory.getLogger(LoadGenerator.class); 

public LoadGenerator(){} 

@Override 
protected void doStart() { 
    executorService = Executors.newFixedThreadPool(2); 
    if (running.compareAndSet(false, true)) { 
     for (int x = 0; x < 10; x++) { 
      executorService.submit(new Producer(Integer.toString(x))); 
     } 
    } 
} 

@Override 
protected void doStop() { 
    if (running.compareAndSet(true, false)) { 
     executorService.shutdown(); 
    } 
} 

protected class Producer implements Runnable { 
    String prefix; 
    Lock lock = new ReentrantLock(); 

    public Producer(String prefix) { 
     this.prefix = prefix; 
    } 

    private void send() { 
     lock.lock(); 
     try{ 
      for(int i = 0; i < 10; i++){ 
       System.out.println(Thread.currentThread().getName() + " " + i); 
      } 
     }finally{ 
      lock.unlock(); 
     } 
    } 

    public void run() { 
     send(); 
    } 
} 
} 

, что я ожидал от него, является список, как

pool-604-thread-1 0 
pool-604-thread-1 1 
pool-604-thread-1 2 
pool-604-thread-1 3 
pool-604-thread-1 4 
pool-604-thread-1 5 
pool-604-thread-1 6 
pool-604-thread-1 7 
pool-604-thread-1 8 
pool-604-thread-1 9 
pool-604-thread-2 0 
pool-604-thread-2 1 
pool-604-thread-2 2 
pool-604-thread-2 3 
pool-604-thread-2 4 
pool-604-thread-2 5 
pool-604-thread-2 6 
pool-604-thread-2 7 
pool-604-thread-2 8 
pool-604-thread-2 9 
... 

Последовательность и порядок в каждом потоке не должны быть прерваны, но факт есть иногда прерывания, такие как:

pool-604-thread-1 0 
pool-604-thread-2 0 
pool-604-thread-2 1 
pool-604-thread-2 2 
pool-604-thread-1 1 
pool-604-thread-1 2 
pool-604-thread-1 3 

Что не так? блокировка работает хорошо в localhost на Eclipse, проверена.

Это потому, что SpringXD является распределенной системой? Но у меня есть только один xdconteier, подключенный к моему xdadmin.

Спасибо.

Answer 1

Вы должны пройти ваш ReentrantLock не создать его в каждом новом Runnable, то Производители должны совместно использовать один ReentrantLock, как:

@Override 
protected void doStart() { 
    executorService = Executors.newFixedThreadPool(2); 
    Lock lock = new ReentrantLock(); // create a share lock to lock in multi threads 
    if (running.compareAndSet(false, true)) { 
     for (int x = 0; x < 10; x++) { 
      executorService.submit(new Producer(Integer.toString(x), lock)); //pass this lock to every producer 
     } 
    } 
} 
protected class Producer implements Runnable { 
    String prefix; 
    final Lock lock; 

    public Producer(String prefix, Lock lock) { 
     this.prefix = prefix; 
     this.lock = lock; 
    } 

    private void send() { 
     lock.lock(); 
     try{ 
      for(int i = 0; i < 10; i++){ 
       System.out.println(Thread.currentThread().getName() + " " + i); 
      } 
     }finally{ 
      lock.unlock(); 
     } 
    } 

    public void run() { 
     send(); 
    } 
}