Thread pool calls blocking methods
<p>I have implemented a queue that stores elements similar to a <a href="https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/BlockingQueue.html" rel="nofollow">BlockingQueue</a>. On retrieval, the consumer may specify a <a href="https://docs.oracle.com/javase/8/docs/api/java/util/function/Predicate.html" rel="nofollow">Predicate</a> that is tested against the queue elements. The queue will return elements in FIFO style, but skip all elements that don't fulfil this predicate. So the returned element may not be the head of the queue. If none of the queue elements fulfil the given predicate, the <code>take()</code>-thread sleeps some time and starts over.</p>

<p>There are a few threads adding elements to this queue and now I need many threads that consume elements from that queue.</p>

<p>Adding elements is simple. But how can I "connect" this queue to a pool of workers (preferably a <a href="https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ThreadPoolExecutor.html" rel="nofollow">ThreadPoolExecutor</a> with dynamic thread management) which retrieve elements from this queue and do some work?</p>

<p>My queue has two methods:</p>

<pre><code>boolean add(E e);
E take(); // blocks
</code></pre>

<p>The queue implementation is basically like the example in <a href="https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/locks/Condition.html" rel="nofollow">Condition</a>, except that it's unbounded and not backed by an array, but a <a href="https://docs.oracle.com/javase/8/docs/api/java/util/LinkedHashSet.html" rel="nofollow">LinkedHashSet</a> which doesn't allow duplicates and maintains insertion order. </p>

<p>I came up with this, but I have no clue whether that's the way to go. Do I really need this extra thread?</p>

<pre><code>SynchronousQueue&lt;Runnable&gt; workQ = new SynchronousQueue&lt;&gt;();
ExecutorService threadPool = new ThreadPoolExecutor(10, 100, 30L, TimeUnit.SECONDS, workQ);
new Thread(() -&gt; {
    try {
        while (true) {
            workQ.put(() -&gt; process(queue.take()));
        }
    } catch (InterruptedException e) {
    }
}).start();
</code></pre>