java - Java 多线程消息传递

Question

我有一个有两个线程的应用程序，一个写入队列，第二个从它异步读取。我需要创建第三个生成 20 多个。新创建的线程将运行直到明确停止。这 20 个线程应该获得“实时”数据以便对其进行分析。20 个中的每一个都有一个唯一的 ID/名称。我需要将相关数据（READ 线程收集）发送到正确的线程（20 个线程）。例如，如果数据包含一个 id（其中）为 2 的字符串 --> 我需要将它发送到 ID = 2 的线程。我的问题是：我应该如何为 20 个线程中的每一个持有一个“指针”并将相关数据发送给它？（我可以在可运行列表中搜索 id（将保存线程）--> 但是我需要调用方法“NewData（string）”以便将数据发送到正在运行的线程）。我该怎么做？TIA 帕兹

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

}