如何分别在 n 个线程中安排一些工作

Question

假设我有 n 个线程同时从共享队列中获取值：

public class WorkerThread implements Runnable{
        private BlockingQueue queue;
        private ArrayList<Integer> counts = new ArrayList<>();
        private int count=0;
        public void run(){
            while(true) {
                queue.pop();
                count++;
            }
        }
}

然后对于每个线程，我想每 5 秒计算一次它有多少项已出队，然后将其存储在自己的列表中（计数） 我在这里看到Print "hello world" every X seconds 如何每 x 秒运行一次代码：

Timer t = new Timer();
t.scheduleAtFixedRate(new TimerTask(){
    @Override
    public void run(){
        counts.add(count);
        count = 0
    }
}, 0, 5000);

问题是我无法访问计数变量和计数列表，除非它们是静态的。但我不希望它们是静态的，因为我不希望不同的线程共享这些变量。

关于如何处理这个问题的任何想法？

Answer 1

我认为不可能为您使用计划执行（Timer 和 ScheduledExecutorService 都不是），因为每个新的计划调用都会使用 while 循环创建一个新任务。所以任务的数量会不断增加。

如果您不需要在运行时访问此计数列表，我会建议这样的：

  static class Task implements Runnable {
    private final ThreadLocal<List<Integer>> counts = ThreadLocal.withInitial(ArrayList::new);
    private volatile List<Integer> result = new ArrayList<>();
    private BlockingQueue<Object> queue;

    public Task(BlockingQueue<Object> queue) {
      this.queue = queue;
    }

    @Override
    public void run() {
      int count = 0;
      long start = System.nanoTime();
      try {
        while (!Thread.currentThread().isInterrupted()) {
          queue.take();
          count++;
          long end = System.nanoTime();
          if ((end - start) >= TimeUnit.SECONDS.toNanos(1)) {
            counts.get().add(count);
            count = 0;
            start = end;
          }
        }
      } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
      }
      // the last value
      counts.get().add(count);
      // copy the result cause it's not possible
      // to access thread local variable outside of this thread
      result = counts.get();
    }

    public List<Integer> getCounts() {
      return result;
    }
  }

  public static void main(String[] args) throws Exception {
    ExecutorService executorService = Executors.newFixedThreadPool(3);
    BlockingQueue<Object> blockingQueue = new LinkedBlockingQueue<>();
    Task t1 = new Task(blockingQueue);
    Task t2 = new Task(blockingQueue);
    Task t3 = new Task(blockingQueue);
    executorService.submit(t1);
    executorService.submit(t2);
    executorService.submit(t3);

    for (int i = 0; i < 50; i++) {
      blockingQueue.add(new Object());
      Thread.sleep(100);
    }
    // unlike shutdown() interrupts running threads
    executorService.shutdownNow();
    executorService.awaitTermination(1, TimeUnit.SECONDS);

    System.out.println("t1 " + t1.getCounts());
    System.out.println("t2 " + t2.getCounts());
    System.out.println("t3 " + t3.getCounts());

    int total = Stream.concat(Stream.concat(t1.getCounts().stream(), t2.getCounts().stream()), t3.getCounts().stream())
        .reduce(0, (a, b) -> a + b);
    // 50 as expected
    System.out.println(total);
  }

Answer 2

• 为什么不是静态 AtomicLong？
• 或者 WorkerThread(s) 可以将他们弹出的消息发布到 TimerTask 或其他地方？ TimerTask 会读取该信息吗？