Java多线程大数组访问

Question

我的主类，根据一些规则生成多个线程。 (20-40 个线程存活很长时间)。 每个线程创建几个线程(短时间)-->我正在使用执行器。 我需要在短时间线程中处理多维数组 --> 我像下面的代码一样编写它 --> 但我认为它效率不高，因为我多次将它传递给这么多线程/任务--.我尝试直接从线程访问它(通过将其声明为公共(public) -> 没有成功) -> 将很高兴获得有关如何改进它的评论/建议。 我还考虑下一步返回一个一维数组作为结果(这可能更好，只是在 Assetfactory 类中更新它)-->，我不知道如何做。 请参阅下面的代码。 谢谢 帕兹

import java.util.concurrent.*;
import java.util.logging.Level;

public class AssetFactory implements Runnable{
    private volatile boolean stop = false;
    private volatile String feed ;
    private double[][][] PeriodRates= new double[10][500][4];

    private String TimeStr,Bid,periodicalRateIndicator;
    private final BlockingQueue<String> workQueue;
    ExecutorService IndicatorPool = Executors.newCachedThreadPool();

    public AssetFactory(BlockingQueue<String> workQueue) {
      this.workQueue = workQueue;
    }   

    @Override
    public void run(){
      while (!stop) {
       try{   
          feed = workQueue.take(); 
            periodicalRateIndicator = CheckPeriod(TimeStr, Bid) ;
            if (periodicalRateIndicator.length() >0) {
                IndicatorPool.submit(new CalcMvg(periodicalRateIndicator,PeriodRates));
            } 

          } 

          if ("Stop".equals(feed)) {
              stop = true ;
          }

       } // try
       catch (InterruptedException ex) {
           logger.log(Level.SEVERE, null, ex);
           stop = true;
       }


      } // while
    } // run  

这是 CalcMVG 类

public class CalcMvg implements Runnable {
    private double [][][] PeriodRates = new double[10][500][4];

    public CalcMvg(String Periods, double[][][] PeriodRates) {
        System.out.println(Periods); 
        this.PeriodRates = PeriodRates ;
    }

    @Override
    public void run(){
       try{
          // do some work with the data of PeriodRates array e.g. print it (no changes to array
          System.out.println(PeriodRates[1][1][1]);
          }
       catch (Exception ex){
          System.out.println(Thread.currentThread().getName() + ex.getMessage());
          logger.log(Level.SEVERE, null, ex);
         }
      }//run

  } // mvg class

Answer 1

这里发生的一些事情似乎是错误的，但由于所提供的代码量有限，很难给出一个好的答案。

首先是实际的编码问题:

• 如果只有一个线程访问某个变量(停止、馈送)，则无需将其定义为 volatile 变量

• 您应该声明仅在该函数的本地上下文(运行方法)中使用的变量，而不是在整个实例(几乎所有变量)中全局使用的变量。这允许 JIT 进行各种优化。

• InterruptedException 应终止线程。因为它是作为终止线程工作的请求抛出的。

• 在您的代码示例中，workQueue 似乎除了让线程进入休眠状态或停止它们之外没有执行任何操作。为什么它不立即为实际工作线程提供所需的工作负载？

然后是代码结构问题:

• 您使用线程来为线程提供工作。这是低效的，因为实际上只有有限数量的核心可以完成工作。由于线程的执行顺序未定义，因此 IndicatorPool 很可能大部分处于空闲状态，或者充满了尚未完成的任务。

• 如果您有一组有限的工作要做，则ExecutorCompletionService可能对您的任务有帮助。

我认为通过重新设计代码结构你将获得最好的速度提升。想象一下以下内容(假设我正确理解你的问题):

• 存在一个由某些数据源(例如文件流、网络)提供的任务阻塞队列。

• 一组等于核心数量的工作线程正在等待该数据源的输入，然后对输入进行处理并将其放入完成队列中。

• 特定的数据集是您工作的“终结者”(例如“null”)。如果线程遇到此终止符，它将完成循环并关闭。

现在，以下内容适用于此构造:

情况一:数据源是瓶颈。它无法通过使用多个线程来加速，因为如果您更频繁地询问，您的硬盘/网络将不会工作得更快。

情况 2:您的计算机上的处理能力是瓶颈，因为您无法处理超出计算机上的工作线程/核心可以处理的数据。

在这两种情况下，结论都是，工作线程需要在准备好处理新数据时立即寻找新数据。因为它们要么需要被搁置，要么需要限制传入的数据。这将确保最大吞吐量。

如果所有工作线程都已终止，则工作完成。这可以是:通过使用 CyclicBarrier 进行跟踪或Phaser类。

工作线程的伪代码:

public void run() {
  DataType e;
  try {
    while ((e = dataSource.next()) != null) {
      process(e);
    }
    barrier.await();
  } catch (InterruptedException ex) {
  }
}

希望这对您的案例有所帮助。