Feeding several threads with data using Java

Question

I have a problem with a part of my code. My program have a thread that is getting input from the keyboard and have several threads that are waiting for that input.

The users selects first to what thread he is going to send that input. So lets says that we have 3 threads (0,1,2) plus the thread that gets the keyboard input. The user will select first what thread he wants to interact with and after that he will send the actual data to that thread.

I have a piece of code that is taking care of that process. I use ´LinkedBlockingQueue´ to achieve it. The keyboard thread puts data in the Queue and the "workers" (the other 3 threads) get that data from that queue.

The problem is that all the threads are listening for that same Queue so I put an ID in that Queue to let the threads know if the data is directed to them or to other thread.

Here is the code:

Thread Thread_OUT = new Thread(new Runnable() {
    
                @SuppressWarnings("unchecked")
                @Override
                public void run() {
                     while(true) {
                    try {
                        Object recibido= sharedQueue.take();
                        sharedQueue.put(recibido);          
                        //System.out.println("Im the thread "+ clientID+" and I got "+recibido.toString());
                        if(Integer.parseInt(recibido.toString())==clientID){ // If it is for me I get the data
                            String x = CommandShellServer.data.get(clientID); // just get the data (it is in a hashmap)
                            CommandShellServer.data.clear(); // empty the hashmap
                            sharedQueue.clear();
                            
                            OUT = do_something(x);
                           
                        }
                        else{ // If it is not I will forward it to other thread
                            Thread.currentThread().wait(100);
//                          sharedQueue.put(recibido);
//                          sharedQueue.clear();
                        }

As you can see in the code what I do is checking if the thread that is handling the information is the one that is directed to If it is, I process it, and if it is no I put that the data again in the queue to let the other threads to check for it.

If I select the thread 0 to interact with it works. If I select others it doesn't.

Answer 1

 /*
  * To change this license header, choose License Headers in Project Properties.
  * To change this template file, choose Tools | Templates
  * and open the template in the editor.
  */
 package Application;

 import java.util.ArrayList;
 import java.util.Scanner;

 /**
 *
  * @author husseyn
 */
 public class producteurConsomateur {

 static Scanner clavier;
 static ArrayList<String> queu;

 public static void main(String[] args) {
    queu=new ArrayList<>();
    new Thread(){

        @Override
        public void run() {
            clavier=new Scanner(System.in);
            while (true) {                    
                try {
                    sleep(1000);
                } catch (Exception e) {
                }
                System.out.print("tape message :");
                String nextLine = clavier.nextLine();
                
                queu.add(nextLine);
               // notifyAll();
                
                
                
            }
        }
        
    }.start();
    
    
    new  Thread(){

        @Override
        public void run() {
        
            while (true) {  
                try {
                    
                try {
                    wait();
                } catch (Exception e) {
                }
                synchronized(this){
                String get = queu.get(0);
                String[] messageFormat = get.split(":");
                String id=messageFormat[0];
                if (id.toLowerCase().equals("id1")) {
                      String message=messageFormat[0];
                      queu.remove(0);
                      System.out.println("message recived to thread ID1 :"+message);
                }}
                } catch (Exception e) {
                }
                
            }
        }
        
    }.start();
    
    new  Thread(){

        @Override
        public void run() {
        
            while (true) {  
                try {
                    
                try {
                    wait();
                } catch (Exception e) {
                }
                synchronized(this){
                String get = queu.get(0);
                String[] messageFormat = get.split(":");
                String id=messageFormat[0];
                if (id.toLowerCase().equals("id3")) {
                      String message=messageFormat[0];
                      queu.remove(0);
                      System.out.println("message recived to thread ID3 :"+message);
                }}
                } catch (Exception e) {
                }
                
            }
        }
        
    }.start();
    
    new  Thread(){

        @Override
        public void run() {
        
            while (true) {  
                try {
                    
                try {
                    wait();
                } catch (Exception e) {
                }
                synchronized(this){
                String get = queu.get(0);
                String[] messageFormat = get.split(":");
                String id=messageFormat[0];
                if (id.toLowerCase().equals("id2")) {
                      String message=messageFormat[0];
                      queu.remove(0);
                      System.out.println("message recived to thread ID2 :"+message);
                }}
                } catch (Exception e) {
                }
                
            }
        }
        
    }.start();
  }

 }

And here I use a shared queue but you have to respect the message format is like id1:hello or id2:lol

Answer 2

Get rid of the shared queue, and let each thread have its own. Then, when you get an input, just dispatch it to the queue of appropriate thread that is intended to receive it.