Fix synchronization
This commit is contained in:
parent
599a2463a3
commit
4a5a42b7fd
2
pom.xml
2
pom.xml
@ -4,7 +4,7 @@
|
||||
<groupId>it.cavallium</groupId>
|
||||
<artifactId>filequeue</artifactId>
|
||||
<name>file queue project</name>
|
||||
<version>3.1.4</version>
|
||||
<version>3.1.5</version>
|
||||
<packaging>jar</packaging>
|
||||
<description>Light weight, high performance, simple, reliable and persistent queue</description>
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
@ -1,17 +1,15 @@
|
||||
package it.cavallium.filequeue;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.concurrent.Semaphore;
|
||||
import java.util.concurrent.atomic.AtomicLong;
|
||||
import java.util.concurrent.locks.LockSupport;
|
||||
|
||||
class QueueToConsumer<T> implements IQueueToConsumer<T> {
|
||||
|
||||
private final long BACKOFF_NS = Duration.ofMillis(2).toNanos();
|
||||
private final long MAX_BACKOFF_NS = Duration.ofMillis(500).toNanos();
|
||||
|
||||
private final Object lock = new Object();
|
||||
private final Semaphore semaphore = new Semaphore(0);
|
||||
private long queued;
|
||||
private final AtomicLong preAddQueued;
|
||||
private final AtomicLong afterAddQueued;
|
||||
private final SimpleQueue<T> queue;
|
||||
private final QueueConsumer<T> consumer;
|
||||
private Manager manager;
|
||||
@ -20,7 +18,8 @@ class QueueToConsumer<T> implements IQueueToConsumer<T> {
|
||||
public QueueToConsumer(SimpleQueue<T> queue, QueueConsumer<T> consumer) {
|
||||
this.queue = queue;
|
||||
this.consumer = consumer;
|
||||
queued = queue.size();
|
||||
this.preAddQueued = new AtomicLong(queue.size());
|
||||
this.afterAddQueued = new AtomicLong(queue.size());
|
||||
}
|
||||
|
||||
public synchronized void startQueue() {
|
||||
@ -34,23 +33,30 @@ class QueueToConsumer<T> implements IQueueToConsumer<T> {
|
||||
@Override
|
||||
public void add(T value) {
|
||||
boolean shouldAdd = true;
|
||||
synchronized (lock) {
|
||||
if (queued == 0 && consumer.tryConsume(value)) {
|
||||
if (preAddQueued.getAndIncrement() == 0) {
|
||||
boolean crashed = true;
|
||||
try {
|
||||
if (consumer.tryConsume(value)) {
|
||||
crashed = false;
|
||||
shouldAdd = false;
|
||||
} else {
|
||||
queued++;
|
||||
preAddQueued.decrementAndGet();
|
||||
}
|
||||
} finally {
|
||||
if (crashed) {
|
||||
shouldAdd = false;
|
||||
preAddQueued.decrementAndGet();
|
||||
}
|
||||
}
|
||||
}
|
||||
if (shouldAdd && !closed) {
|
||||
queue.add(value);
|
||||
semaphore.release();
|
||||
afterAddQueued.incrementAndGet();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
closed = true;
|
||||
semaphore.release();
|
||||
}
|
||||
|
||||
private class Manager extends Thread {
|
||||
@ -59,15 +65,10 @@ class QueueToConsumer<T> implements IQueueToConsumer<T> {
|
||||
public void run() {
|
||||
try {
|
||||
while (!closed) {
|
||||
boolean shouldRemove = false;
|
||||
T element;
|
||||
synchronized (lock) {
|
||||
if (queued > 0) {
|
||||
queued--;
|
||||
shouldRemove = true;
|
||||
}
|
||||
}
|
||||
semaphore.acquire();
|
||||
boolean shouldRemove = preAddQueued.getAndUpdate(n -> n > 0 ? n - 1 : 0) > 0;
|
||||
while (!closed) {
|
||||
if (afterAddQueued.get() > 0) {
|
||||
if (!closed && shouldRemove) {
|
||||
element = queue.remove();
|
||||
long nextDelay = BACKOFF_NS;
|
||||
@ -81,7 +82,9 @@ class QueueToConsumer<T> implements IQueueToConsumer<T> {
|
||||
}
|
||||
}
|
||||
}
|
||||
} catch (InterruptedException ex) {
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
closed = true;
|
||||
}
|
||||
}
|
||||
|
Loading…
Reference in New Issue
Block a user