java - 发送到 ExecutorService 的作业的运行时间

Question

你好，

我正在编写一个程序，其中为从文本文件中读取的每一行调用一个方法。由于此方法的每次调用都独立于读取的任何其他行，因此我可以并行调用它们。为了最大限度地提高 CPU 使用率，我使用了一个 ExecutorService，我在其中提交了每个 run() 调用。由于文本文件有 1500 万行，我需要错开 ExecutorService 运行，以免一次创建太多作业(OutOfMemory 异常)。我还想跟踪每次提交的运行运行的时间，因为我看到有些运行没有完成。问题是，当我尝试使用带超时的 Future.get 方法时，超时指的是自它进入 ExecutorService 队列以来的时间，而不是自它开始运行以来的时间，如果它甚至开始的话。我想获取它开始运行以来的时间，而不是它进入队列后的时间。

代码如下所示:

ExecutorService executorService= Executors.newFixedThreadPool(ncpu);
line = reader.readLine();
long start = System.currentTimeMillis();
HashMap<MyFut,String> runs = new HashMap<MyFut, String>();
HashMap<Future, MyFut> tasks = new HashMap<Future, MyFut>();
while ( (line = reader.readLine()) != null ) { 

String s = line.split("\t")[1];
final String m = line.split("\t")[0];
MyFut f = new MyFut(s, m);
tasks.put(executorService.submit(f), f);

runs.put(f, line);

while (tasks.size()>ncpu*100){
    try {
        Thread.sleep(100);
    } catch (InterruptedException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    }

    Iterator<Future> i = tasks.keySet().iterator();
    while(i.hasNext()){
        Future task = i.next();
        if (task.isDone()){
            i.remove();

        } else {
            MyFut fut = tasks.get(task);
            if (fut.elapsed()>10000){
                System.out.println(line);
                task.cancel(true);
                i.remove();
            }
        }
    }
}
}

private static class MyFut implements Runnable{

private long start;
String copy;
String id2;

public MyFut(String m, String id){
    super();

    copy=m;
    id2 = id;
}

public long elapsed(){
    return System.currentTimeMillis()-start;
}



@Override
public void run() {
    start = System.currentTimeMillis();
    do something...
}

}

如您所见，我会尝试跟踪已发送的作业数量，如果超过阈值，我会稍等片刻，直到一些作业完成。我还尝试检查是否有任何作业取消它的时间太长，记住哪个作业失败了，然后继续执行。这没有像我希望的那样工作。一项任务执行 10 秒远远超过所需(我在 70 到 130 秒内完成了 1000 行，具体取决于机器和 CPU 数量)。

我做错了什么？难道我的 Runnable 类中的 run 方法只在 ExecutorService 中的某个线程空闲并开始处理它时才被调用吗？我得到了很多需要 10 秒以上的结果。有没有更好的方法来实现我正在尝试的目标？

谢谢。

Answer 1

如果您使用的是 Future，我建议将 Runnable 更改为 Callable，并返回执行线程的总时间作为结果。下面是示例代码:

import java.util.concurrent.Callable;

public class MyFut implements Callable<Long> {

    String copy;
    String id2;

    public MyFut(String m, String id) {
        super();

        copy = m;
        id2 = id;
    }

    @Override
    public Long call() throws Exception {
        long start = System.currentTimeMillis();
        //do something...
        long end = System.currentTimeMillis();
        return (end - start);
    }
}