为什么 ReentrantLock 在 SpringXD 中不起作用

Question

我正在尝试在 SpringXD 上的工作中添加 ReentrantLock，但有时似乎存在跨线程中断。

例如，我有这些代码：

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

怎么了？该锁在 Eclipse 的 localhost 中运行良好，经过测试。

是不是因为SpringXD是分布式系统？但是我只有一个 xdcontaier 连接到我的 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();
    }
}