Extending ThreadPoolExecutor I have implemented a ThreadPoolExecutor that will run a Consumer&lt;T&gt; only on elements not already consumed. This code uses Java 8.

The background behind this is that I scan a directory every x time units for which files are present, I must maintain a 100% accuracy on finding files and other mechanisms such as JNotify or a simple plain WatcherService do not achieve that.

public class SingleExecutionThreadPoolExecutor&lt;E&gt; extends ThreadPoolExecutor {
    private final Consumer&lt;E&gt; consumer;
    private final List&lt;E&gt; elementsInProcess = new ArrayList&lt;&gt;();

    public SingleExecutionThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue&lt;Runnable&gt; workQueue, Consumer&lt;E&gt; consumer) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
        this.consumer = consumer;
    }

    public SingleExecutionThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue&lt;Runnable&gt; workQueue, Consumer&lt;E&gt; consumer, RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
        this.consumer = consumer;
    }

    public SingleExecutionThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue&lt;Runnable&gt; workQueue, Consumer&lt;E&gt; consumer, ThreadFactory threadFactory) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
        this.consumer = consumer;
    }

    public SingleExecutionThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue&lt;Runnable&gt; workQueue, Consumer&lt;E&gt; consumer, ThreadFactory threadFactory, RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
        this.consumer = consumer;
    }

    public void execute(E element) {
        if (!elementsInProcess.contains(element)) {
            super.execute(() -&gt; consumer.accept(element));
            elementsInProcess.add(element);
        }
    }

    @Override
    @Deprecated
    public void execute(Runnable command) { }
}


It is called using the following classes and snippets:

public final class UniqueTimePath {
    private final Path path;
    private final FileTime fileTime;

    public UniqueTimePath(final Path path) {
        this.path = path;
        try {
            this.fileTime = Files.getLastModifiedTime(path);
        } catch (IOException ex) {
            throw new UncheckedIOException(ex);
        }
    }

    public Path getPath() {
        return path;
    }

    public FileTime getFileTime() {
        return fileTime;
    }

    @Override
    public int hashCode() {
        int hash = 3;
        return hash;
    }

    @Override
    public boolean equals(Object obj) {
        if (obj == null) {
            return false;
        }
        if (getClass() != obj.getClass()) {
            return false;
        }
        final UniqueTimePath other = (UniqueTimePath) obj;
        if (!Objects.equals(this.path, other.path)) {
            return false;
        }
        if (!Objects.equals(this.fileTime, other.fileTime)) {
            return false;
        }
        return true;
    }

    @Override
    public String toString() {
        return "{" + path + ", " + fileTime + "}";
    }
}




private SingleExecutionThreadPoolExecutor&lt;UniqueTimePath&gt; executor;
//...
executor = new SingleExecutionThreadPoolExecutor&lt;&gt;(threadCount, threadCount, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue&lt;&gt;(), fileConsumer);
//...
Files.list(directory)
        .map(UniqueTimePath::new)
        .forEach(executor::execute);
.equals

The last part of your .equals implementation can be simplified to:

return Objects.equals(this.path, other.path) &amp;&amp; Objects.equals(this.fileTime, other.fileTime);


HashCode

int hash = 3;
return hash;


Do I even have to comment anything about what I think about this implementation? You use Objects.equals above, you can use Objects.hashCode(this.path, this.fileTime); here. It is preferred if hashCode and equals use the same fields for their operations.

Set or List?

Whenever you use a List in a program, ask yourself the question: Does the order of the elements matter?

In this case, they don't. Use the Set interface and create as HashSet for your elementsInProcess variable.

private final Set&lt;E&gt; elementsInProcess = new HashSet&lt;&gt;();


When using a Set, it is extra important that the elements in the set has properly implemented hashCode and equals! (Or are using the default implementations inherited from the Object class)

Cleanup

I suggest you override the shutdown methods of a ThreadPoolExecutor to clear the elementsInProcess collection.

execute(Runnable command)

Please throw a UnsupportedOperationException or similar in this method:

public void execute(Runnable command) { }


Edit: Scratch that. I think palacsint is absolutely right. By doing nothing here, you're breaking the contract of the Executor interface (which you implement, whether you know it or not - by extending an ExecutorService). The Javadoc for the execute(Runnable command) method says: 


  Executes the given command at some time in the future.


As it currently stands, you are not doing that and have no real ability of doing so in that method. This fact is an indication that you should not use extends ThreadPoolExecutor and instead use a private ThreadPoolExecutor executor;, and therefore using composition over inheritance.

Concurrency

If two threads would call execute at the same time, there is a concurrency issue on elementsInProcess. You need to synchronize on something there, for minimal blocking I suggest you do something like this:

synchronize (lock) {
    if (elementsInProcess.contains(element)) {
        return;
    }
    elementsInProcess.add(element);
}
super.execute(() -&gt; consumer.accept(element));


Where lock is a private final Object lock = new Object();

Or, you could utilize the fact that the .add method of a set actually returns a boolean indicating if the add was successful and use a Collections.synchronizedSet.

if (!elementsInProcess.add(element)) {
    return;
}
super.execute(() -&gt; consumer.accept(element));


It is important when using a synchronized Set that you only do one method call here, as there could otherwise be a gap between .contains and .add which would again cause concurrency issues between threads.