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); 
    } 
    } 

}