SetBlockingQueue.java

package xxx;

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

import java.util.Collections;
import java.util.Set;
import java.util.concurrent.*;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Consumer;

@Slf4j
public class SetBlockingQueue<T> extends LinkedBlockingQueue<T> {

    private final Set<T> set = Collections.newSetFromMap(new ConcurrentHashMap<>());
    private final ReentrantLock addLock = new ReentrantLock();
    private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();

    private final ScheduledFuture<?> scheduledFuture = scheduledExecutorService.scheduleAtFixedRate(
            () -> log.info("Queue size is " + set.size()),
            5L,
            1L,
            TimeUnit.SECONDS);

    @Override
    public boolean add(T t) {
        ReentrantLock lock = this.addLock;
        lock.lock();
        try {
            if (set.contains(t)) {
                return false;
            } else {
                set.add(t);
                return super.add(t);
            }
        } finally {
            lock.unlock();
        }
    }

    @Override
    @SneakyThrows
    public T take()  {
        T t = super.take();
        set.remove(t);
        return t;
    }

    public boolean isEmpty() {
        return set.isEmpty() && super.isEmpty();
    }

    @SneakyThrows
    public synchronized void clear() {
        super.clear();
        set.clear();
        scheduledFuture.cancel(true);
    }

    public int size() {
        return set.size();
    }

    public void forEach(Consumer<? super T> action) {
        super.forEach(action);
    }
}