Java – the queue is completely wrong. Cooperate with multiple consumers and producers

I want to simulate the following scenario: multiple users and producer threads are modifying some data

establish

BlockingQueue<String> q1 = new SynchronousQueue<String>();
    BlockingQueue<String> q2 = new SynchronousQueue<String>();

    Producer dataProducer = new Producer(q1); // publish to q1

    Filter1 filter1 = new Filter1(q1,q2);    // read from q1,publish to q2
    Filter2 filter2 = new Filter2(q2);        // read from q2

    new Thread(dataProducer,"Producer-Thread").start();
    new Thread(filter1,"Filter1-Thread").start();
    new Thread(filter2,"Filter2-Thread").start();

film producer

public void run() {
    try {
        while (true) {
            this.q.put(saySomething());
        }
    } catch (InterruptedException ex) {
        ex.printStackTrace();
    }
}

public String saySomething() {
    return "Something";
}

Filter 1

public void run() {
    try {
        while (true) {
            consume(qIn.take());
        }
    } catch (InterruptedException ex) {
        ex.printStackTrace();
    }
}

private void consume(String take) {
    //Modify data according to some rules
    String newData = take.replace("m","-");
    produce(newData);
}

private void produce(String newData) {
    // put new data in queue out
    qOut.add(newData);                   // <-- Stacktrace points here
}

Filter 2

public void run() {
    try {
        while (true) {
            consume(qIn.take());
        }
    } catch (InterruptedException ex) {
        ex.printStackTrace();
    }
}

private void consume(String s) {
    System.out.println("Something became: " + s);
}

So, recall that producer puts some content into the queue read by filter 1 It modifies the data and publishes it to another queue read by filter 2 Filter 2 prints the final data

This code failed

Exception in thread "Thread-2" java.lang.IllegalStateException: Queue full

Can you help me understand why?

Solution

You should use put () instead of add (). A synchronousqueue is always full and empty. It has no depth Put () tells the queue to pause this thread until another thread enters to take the element out of the queue

The add () method will succeed only when the thread is waiting. If there is no thread waiting, you will get a queue full exception