Java多線程消息傳遞

Question

我有兩個線程的應用程序，一個寫入隊列，另一個讀取異步。 我需要創建第三個生成20多個。 新創建的線程將運行直到明確停止。那20個線程應該獲得「實時」數據以分析它。 每個20都有一個唯一的ID /名稱。我需要將相關數據（即READ線程收集）發送到正確的線程（20個線程）。例如如果數據包含一個id爲2的字符串 - >我需要將它發送到ID = 2的線程。 我的問題是：我應該如何爲20個線程中的每一個持有一個「指針」並將相關數據發送給它？ （我可以搜索可運行列表中的id（這將保持線程） - >但是然後我需要調用一個方法「NewData（string）」來將數據發送到正在運行的線程）。 我該怎麼做？ TIA Paz

Answer 1

您可能會更好地使用隊列與您的線程進行通信。然後，您可以將所有隊列放入地圖中以便於訪問。我會推薦一個BlockingQueue。

public class Test { 
    // Special stop message to tell the worker to stop. 
    public static final Message Stop = new Message("Stop!"); 

    static class Message { 
    final String msg; 

    // A message to a worker. 
    public Message(String msg) { 
     this.msg = msg; 
    } 

    public String toString() { 
     return msg; 
    } 

    } 

    class Worker implements Runnable { 
    private volatile boolean stop = false; 
    private final BlockingQueue<Message> workQueue; 

    public Worker(BlockingQueue<Message> workQueue) { 
     this.workQueue = workQueue; 
    } 

    @Override 
    public void run() { 
     while (!stop) { 
     try { 
      Message msg = workQueue.poll(10, TimeUnit.SECONDS); 
      // Handle the message ... 

      System.out.println("Worker " + Thread.currentThread().getName() + " got message " + msg); 
      // Is it my special stop message. 
      if (msg == Stop) { 
      stop = true; 
      } 
     } catch (InterruptedException ex) { 
      // Just stop on interrupt. 
      stop = true; 
     } 
     } 
    } 
    } 

    Map<Integer, BlockingQueue<Message>> queues = new HashMap<>(); 

    public void test() throws InterruptedException { 
    // Keep track of my threads. 
    List<Thread> threads = new ArrayList<>(); 
    for (int i = 0; i < 20; i++) { 
     // Make the queue for it. 
     BlockingQueue<Message> queue = new ArrayBlockingQueue(10); 
     // Build its thread, handing it the queue to use. 
     Thread thread = new Thread(new Worker(queue), "Worker-" + i); 
     threads.add(thread); 
     // Store the queue in the map. 
     queues.put(i, queue); 
     // Start the process. 
     thread.start(); 
    } 

    // Test one. 
    queues.get(5).put(new Message("Hello")); 

    // Close down. 
    for (BlockingQueue<Message> q : queues.values()) { 
     // Stop each queue. 
     q.put(Stop); 
    } 

    // Join all threads to wait for them to finish. 
    for (Thread t : threads) { 
     t.join(); 
    } 
    } 

    public static void main(String args[]) { 
    try { 
     new Test().test(); 
    } catch (Throwable t) { 
     t.printStackTrace(System.err); 
    } 
    } 

}