Fix synchronization
This commit is contained in:
parent
477932e051
commit
599a2463a3
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.3</version>
|
||||
<version>3.1.4</version>
|
||||
<packaging>jar</packaging>
|
||||
<description>Light weight, high performance, simple, reliable and persistent queue</description>
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
@ -22,23 +22,17 @@ public final class LMDBQueueToConsumer<T> implements IQueueToConsumer<T> {
|
||||
|
||||
@Override
|
||||
public void add(T value) {
|
||||
synchronized (queue) {
|
||||
queue.add(value);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
synchronized (queue) {
|
||||
queue.close();
|
||||
queueLMDB.close();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void startQueue() {
|
||||
synchronized (queue) {
|
||||
queue.startQueue();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
@ -42,9 +42,7 @@ class QueueToConsumer<T> implements IQueueToConsumer<T> {
|
||||
}
|
||||
}
|
||||
if (shouldAdd && !closed) {
|
||||
synchronized (queue) {
|
||||
queue.add(value);
|
||||
}
|
||||
semaphore.release();
|
||||
}
|
||||
}
|
||||
@ -71,9 +69,7 @@ class QueueToConsumer<T> implements IQueueToConsumer<T> {
|
||||
}
|
||||
semaphore.acquire();
|
||||
if (!closed && shouldRemove) {
|
||||
synchronized (queue) {
|
||||
element = queue.remove();
|
||||
}
|
||||
long nextDelay = BACKOFF_NS;
|
||||
while (!closed && !consumer.tryConsume(element)) {
|
||||
LockSupport.parkNanos(nextDelay);
|
||||
|
@ -1,5 +1,8 @@
|
||||
package it.cavallium.filequeue;
|
||||
|
||||
/**
|
||||
* The queue must be thread-safe
|
||||
*/
|
||||
interface SimpleQueue<T> {
|
||||
|
||||
void add(T element);
|
||||
|
@ -17,7 +17,7 @@ class SimpleQueueFile<T> implements SimpleQueue<T> {
|
||||
}
|
||||
|
||||
@Override
|
||||
public void add(T element) {
|
||||
public synchronized void add(T element) {
|
||||
try {
|
||||
queueFile.add(ser.serialize(element));
|
||||
} catch (IOException e) {
|
||||
@ -26,7 +26,7 @@ class SimpleQueueFile<T> implements SimpleQueue<T> {
|
||||
}
|
||||
|
||||
@Override
|
||||
public T remove() {
|
||||
public synchronized T remove() {
|
||||
try {
|
||||
byte[] element = queueFile.peek();
|
||||
if (element == null) {
|
||||
@ -41,7 +41,7 @@ class SimpleQueueFile<T> implements SimpleQueue<T> {
|
||||
}
|
||||
|
||||
@Override
|
||||
public int size() {
|
||||
public synchronized int size() {
|
||||
return queueFile.size();
|
||||
}
|
||||
}
|
||||
|
@ -2,10 +2,16 @@ package it.cavallium.filequeue;
|
||||
|
||||
import java.util.Queue;
|
||||
|
||||
/**
|
||||
* Thread safe queue
|
||||
*/
|
||||
class SimpleQueueJava<T> implements SimpleQueue<T> {
|
||||
|
||||
private final Queue<T> queue;
|
||||
|
||||
/**
|
||||
* @param queue the queue must be thread-safe
|
||||
*/
|
||||
public SimpleQueueJava(Queue<T> queue) {
|
||||
this.queue = queue;
|
||||
}
|
||||
|
@ -13,6 +13,9 @@ import org.lmdbjava.DbiFlags;
|
||||
import org.lmdbjava.Env;
|
||||
import org.lmdbjava.Txn;
|
||||
|
||||
/**
|
||||
* Thread safe queue
|
||||
*/
|
||||
public class SimpleQueueLMDB<T> implements SimpleQueue<T>, Closeable {
|
||||
|
||||
private static final Set<String> REGISTRY = new HashSet<>();
|
||||
|
Loading…
Reference in New Issue
Block a user