Question about BlockingQueue

[Peter Curila]

Could somebody explain me why the "take" message is first and then is "put" when I run the test method test100().
It works but the messages confuse me.

I have class NumberQueue which is simple LinkedBlockingQueue for Integers with some messages after/before put and take.
Then I have test case class NumberQueueTest where I have defined two nested classes Filler and Taker. Filler takes array of int
and put them into the queue and Taker takes them from the queue. Boths classes extends Thread of course. In test test100 i create 2Fillers and 2Takers.
Its just test so it does not do anything meaningful.

Here is the output

Queue constructed
taker 0 before take:
taker 1 before take:
filler 0 before put: 1
taker 0 after take: 1 // here I took 1
taker 0 before take:
filler 0 after put: 1 // and here I put 1
filler 0 before put: 2
taker 1 after take: 2
taker 1 before take:
filler 0 after put: 2
filler 1 before put: 3
taker 0 after take: 3
filler 1 after put: 3
filler 1 before put: 4
taker 1 after take: 4
filler 1 after put: 4

and here is the code:

  
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
 
public class NumberQueue {
 
    BlockingQueue<Integer> queue;
 
    public NumberQueue(int capacity) {
        queue = new LinkedBlockingQueue<Integer>(capacity);
        System.out.println("Queue constructed");
    }
 
    public void put(int number) throws InterruptedException {
        System.out.println(Thread.currentThread().getName() + " before put:  " + number);
        queue.put(number);
        System.out.println(Thread.currentThread().getName() + " after  put:  " + number);
    }
 
    public int take() throws InterruptedException {
        System.out.println(Thread.currentThread().getName() + " before take: ");
        int result = queue.take();
        System.out.println(Thread.currentThread().getName() + " after  take: " + result);
        return result;
    }
}
 
import java.util.*;
import junit.framework.TestCase;
 
public class NumberQueueTest extends TestCase {
 
    public NumberQueueTest(String sTestName) {
        super(sTestName);
    }
 
    public void test100() {
        NumberQueue queue = new NumberQueue(3);
        int currentNo = 0;
        final int arraySize = 2;
        final int threadCount = 2;
        int[][] arrayToFill = new int[threadCount][arraySize];
        Filler[] filler = new Filler[threadCount];
        Taker[] taker = new Taker[threadCount];
 
        for (int i = 0; i < filler.length; i++) {
            for (int j = 0; j < arraySize; j++) {
                arrayToFill[i][j] = ++currentNo;
            }
            filler[i] = new Filler(queue, arrayToFill[i], i);
            taker[i] = new Taker(queue, arrayToFill[i].length, i);
        }
 
        for (int i = 0; i < taker.length; i++) {
            taker[i].start();
        }
 
        for (int i = 0; i < filler.length; i++) {
            filler[i].start();
        }
 
        try {
            Thread.sleep(1000);
        } catch (InterruptedException ie) {
            ie.printStackTrace();
        }
 
        Collection<Integer> results = new HashSet<Integer>(threadCount * arraySize);
        int[] takerResult = new int[arraySize];
        for (int i = 0; i < taker.length; i++) {
            takerResult = taker[i].getResults();
            for (int j = 0; j < takerResult.length; j++) {
                results.add(takerResult[j]);
            }
        }
        assertEquals( threadCount * arraySize, results.size());
    }
 
    class Filler extends Thread {
 
        NumberQueue queue;
        int[] numbers;
 
        Filler(NumberQueue queue, int[] numbers, int threadNo) {
            super("filler " + threadNo);
            this.queue = queue;
            this.numbers = numbers;
        }
 
        public void run() {
            try {
                for (int i = 0; i < numbers.length; i++) {
                    queue.put(numbers[i]);
                }
            } catch (InterruptedException ie) {
                ie.printStackTrace();
            }
        }
    }
 
    class Taker extends Thread {
 
        NumberQueue queue;
        int[] results;
        boolean done = false;
 
        Taker(NumberQueue queue, int arrayLength, int threadNo) {
            super("taker  " + threadNo);
            this.queue = queue;
            this.results = new int[arrayLength];
        }
 
        public void run() {
            try {
                for (int i = 0; i < results.length; i++) {
                    results[i] = queue.take();
                }
                done = true;
            } catch (InterruptedException ie) {
                ie.printStackTrace();
            }
        }
 
        public int[] getResults() {
            while (!done) {
            }
            return results;
        }
    }
}
 

[Joe Ess]

Your put() and take() methods are not synchronized so they do not perform their actions atomically (i.e. without interruption). I haven't looked at BlockingQueue, but when you "put" something onto the BlockingQueue, it probably invokes notify() or notifyAll() to wake threads waiting on get() invocations. The thread that invoked take() gets the CPU at that point and you see the print outs from that execution.