Referencing a Producer Thread from a Consumer Thread

Question

I'm currently developing a programme using multiple producer threads, and one consumer thread. I am wondering if there is an easy way to reference exactly which Producer Thread the Consumer Thread has consumed from.

Here's an example of my output so far:

ConsumerThread consumed: 12 bytes

I would like it to be, for example:

ConsumerThread consumed: 12 bytes from ThreadA

ConsumerThread consumed: 62 bytes from ThreadB

Here is my Consumer Code, called CPU in this case:

class CPU implements Runnable {

private final Vector processingQueue;
private final int SIZE;

public CPU (Vector processingQueue, int size) {
    this.processingQueue = processingQueue;
    this.SIZE = size;
}

public void run() {
    while (true) {
        try {
            System.out.println("CPU processing: " + consume() + " bytes");
            Thread.sleep(50);
        } catch (InterruptedException ex) {
            Logger.getLogger(CPU.class.getName()).log(Level.SEVERE, null, ex);
        }

    }
}

private int consume() throws InterruptedException {
    //wait if queue is empty
    while (processingQueue.isEmpty()) {

        synchronized (processingQueue) {
            System.out.println("Queue is empty " + Thread.currentThread().getName()
                    + " is waiting , size: " + processingQueue.size());

            processingQueue.wait();
        }
    }
    //Otherwise consume element and notify waiting producer
    synchronized (processingQueue) {
        processingQueue.notifyAll();
        return (Integer) processingQueue.remove(0);
    }
}
}

Here's an example of one of my producers, called OperatingSystem:

public class OperatingSystem extends Thread {

private final Vector processingQueue;
private final int SIZE;

public OperatingSystem (Vector processingQueue, int size) {
    this.processingQueue = processingQueue;
    this.SIZE = size;
}

private void produce(int i) throws InterruptedException {

    // suspend producing if queue is full
    while (processingQueue.size() == SIZE) {

        synchronized (processingQueue) {
            System.out.println("Queue is full " + Thread.currentThread().getName()
                    + " is waiting , size: " + processingQueue.size());

            processingQueue.wait();
        }
    }

    // producing element and notify consumers
    synchronized (processingQueue) {
        processingQueue.add(i);
        processingQueue.notifyAll();
    }
}

public void run() {
    //OperatingSystem using 300bytes
    for (int i = 0; i <= 300; i++) {
        System.out.println("Operating System producing: " + i + " bytes");
        try {
            produce(i);
        } catch (InterruptedException ex) {
            Logger.getLogger(OperatingSystem.class.getName()).log(Level.SEVERE, null, ex);
        }
    }
}
}//OperatingSystem

Any help would be great, thanks!

Answer 1

Each producer would have to attach its name / identifier to the queue with the actual information (in your case, an int).

class Data {
    int data;
    String source;
}

And, instead of writing and reading integers from the queue, use Data instances.