Beginner Java Multithreading issue

I've had this problem before, and the only way I found was to send a special marker object to the BlockingQueue. When the Queue .take() the object, if this is the marker, then the Thread ends itself.

I've tried other solutions, like to wake up the thread and detect the Exception, with no success.

There's a pattern called the Poison Pill that's good for this. Basically, when the producers are done, insert a special value into the queue that tells a consumer to stop. You can either insert one pill for each consumer, or, once a consumer gets a poison pill, return it to the queue for the next consumer. Since it sounds like you're just enqueuing strings, something like

public static final String POISON_PILL = "DONE";

Or in Java 8, use Optional to wrap your values, then have not present be the pill.

BlockingQueue<Optional<...>> queue;

Another option is using an ExecutorService (which is actually backed by a BlockingQueue) and submitting each file as its own task, then using executorService.shutdown() when you're done. The problem with this is that it couples your code more tightly than needed, and it makes it harder to reuse resources like database and HTTP connections.

I'd avoid interrupting your workers to signal them because that can cause blocking IO operations to fail.

You might use the approch below. Add observer pattern if you need to. Or simply - instead of signalling with a death packet, collect a list of waiting Threads and then interrupt() them.

public class AccessCountingLinkedPrioQueue<T> {

    private final LinkedBlockingQueue<T>    mWrappingQueue                  = new LinkedBlockingQueue<>();
    private final Object                    mSyncLockObj                    = new Object();

    private final int                       mMaxBlockingThreads;
    private final T                         mDeathSignallingObject;

    private volatile int                    mNumberOfThreadsInAccessLoop    = 0;

    public AccessCountingLinkedPrioQueue(final int pMaxBlockingThreads, final T pDeathSignallingObject) {
        mMaxBlockingThreads = pMaxBlockingThreads;
        mDeathSignallingObject = pDeathSignallingObject;
    }

    public T take() throws InterruptedException {
        final T retVal;
        synchronized (mSyncLockObj) {
            ++mNumberOfThreadsInAccessLoop;
        }
        synchronized (mWrappingQueue) {
            if (mNumberOfThreadsInAccessLoop >= mMaxBlockingThreads && mWrappingQueue.isEmpty()) signalDeath();
            retVal = mWrappingQueue.take();
        }
        synchronized (mSyncLockObj) {
            --mNumberOfThreadsInAccessLoop;
        }
        return retVal;
    }

    private void signalDeath() {
        for (int i = 0; i < mMaxBlockingThreads; i++) {
            mWrappingQueue.add(mDeathSignallingObject);
        }
    }

    public int getNumberOfThreadsInAccessLoop() {
        return mNumberOfThreadsInAccessLoop;
    }
}

class WorkPacket {
    // ... your content here
}

class MultiThreadingBoss {
    static public final WorkPacket  DEATH_FROM_ABOVE    = new WorkPacket();

    public MultiThreadingBoss() {
        final int THREADS = 7;
        final AccessCountingLinkedPrioQueue<WorkPacket> prioQ = new AccessCountingLinkedPrioQueue<>(THREADS, DEATH_FROM_ABOVE);
        for (int i = 0; i < THREADS; i++) {
            final ThreadedWorker w = new ThreadedWorker(prioQ);
            new Thread(w).start();
        }
    }
}

class ThreadedWorker implements Runnable {
    private final AccessCountingLinkedPrioQueue<WorkPacket> mPrioQ;

    public ThreadedWorker(final AccessCountingLinkedPrioQueue<WorkPacket> pPrioQ) {
        mPrioQ = pPrioQ;
    }

    @Override public void run() {
        while (true) {
            try {
                final WorkPacket p = mPrioQ.take();
                if (p == MultiThreadingBoss.DEATH_FROM_ABOVE) break; // or return

                // ... do your normal work here

            } catch (final InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}