diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/ThrottleLimitResultQueue.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/ThrottleLimitResultQueue.java index b391e456e..f8edcd09e 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/ThrottleLimitResultQueue.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/ThrottleLimitResultQueue.java @@ -19,7 +19,6 @@ package org.springframework.batch.repeat.support; import java.util.NoSuchElementException; import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingQueue; -import java.util.concurrent.Semaphore; /** * An implementation of the {@link ResultQueue} that throttles the number of @@ -33,19 +32,21 @@ public class ThrottleLimitResultQueue implements ResultQueue { private final BlockingQueue results; // Accumulation of dummy objects flagging expected results in the future. - private final Semaphore waits; + private volatile int waits = 0; private final Object lock = new Object(); private volatile int count = 0; + private final int throttleLimit; + /** * @param throttleLimit the maximum number of results that can be expected * at any given time. */ public ThrottleLimitResultQueue(int throttleLimit) { + this.throttleLimit = throttleLimit; results = new LinkedBlockingQueue(); - waits = new Semaphore(throttleLimit); } public boolean isEmpty() { @@ -60,7 +61,7 @@ public class ThrottleLimitResultQueue implements ResultQueue { public boolean isExpecting() { // Base the decision about whether we expect more results on a // counter of the number of expected results actually collected. - // Do not synchronize! Otherwise put and expect can deadlock. + // Do not synchronize! Otherwise put and expect can deadlock. return count > 0; } @@ -73,7 +74,10 @@ public class ThrottleLimitResultQueue implements ResultQueue { */ public void expect() throws InterruptedException { synchronized (lock) { - waits.acquire(); + while (waits >= throttleLimit) { + lock.wait(); + } + waits++; count++; } } @@ -84,9 +88,12 @@ public class ThrottleLimitResultQueue implements ResultQueue { } // There should be no need to block here, or to use offer() results.add(holder); - // Take from the waits queue now to allow another result to + // Take from the waits now to allow another result to // accumulate. But don't decrement the counter. - waits.release(); + synchronized (lock) { + waits--; + lock.notifyAll(); + } } public T take() throws NoSuchElementException, InterruptedException {