使用 Callable 和 ExecutorCompletionService，future.cancel() 不起作用

Question

我正在使用一堆可调用对象来搜索单个块中的列表，一旦返回 true，我想取消所有其他正在运行的可调用对象。 future.cancel 不会取消它们

我的号码查找器

    public class NumberFinderImpl implements NumberFinder {
  // how many threads the exucutor will use to start the callables
  private final int NUMBER_THREADS = 20;
  // the amount of elements in the array we will search in each callable chunk
  private final int CHUNK_ARRAY_SIZE = 5;

  @Override
  public boolean contains(int valueToFind, List<CustomNumberEntity> arrayToSearch) {
    long startTime = System.nanoTime();
    ExecutorService WORKER_THREAD_POOL = Executors.newFixedThreadPool(NUMBER_THREADS);
    CompletionService<Boolean> completionService =
        new ExecutorCompletionService<>(WORKER_THREAD_POOL);
    int numberOfChunksNeeded = (int) Math.ceil(arrayToSearch.size() / CHUNK_ARRAY_SIZE);
    // get a callable for each chunk search
    List<Callable<Boolean>> callablesForEachChunkSearch =
        getCallablesForEachChunk(
            CHUNK_ARRAY_SIZE, numberOfChunksNeeded, valueToFind, arrayToSearch);
    // start the callables and collect the futures
    List<Future<Boolean>> futuresForCallables =
        callablesForEachChunkSearch
            .stream()
            .map(completionService::submit)
            .collect(Collectors.toList());
    for (int j = 0; j < futuresForCallables.size(); j++) {
      try {
        // take().get() is blocking
        // so if a callable is not done yet
        // it will wait until it is before proceeding
        Boolean chunkResult = completionService.take().get();
        if (chunkResult) {
          long endTime = System.nanoTime();
          long timeTaken = endTime - startTime;
          // TimeUnit
          long timeInSeconds = TimeUnit.SECONDS.convert(timeTaken, TimeUnit.NANOSECONDS);
          System.out.println("Search time in seconds" + timeInSeconds);
          for (Future<Boolean> future : futuresForCallables) {
            // cancel all the other running callables
            future.cancel(true);
          }
          return true;
        }
      } catch (InterruptedException e) {
        e.printStackTrace();
      } catch (ExecutionException e) {
        e.printStackTrace();
      }
    }
    for (Future<Boolean> future : futuresForCallables) {
      // cancel all the other running callables
      future.cancel(true);
    }
    long endTime = System.nanoTime();
    long timeTaken = endTime - startTime;
    // TimeUnit
    long timeInSeconds = TimeUnit.SECONDS.convert(timeTaken, TimeUnit.NANOSECONDS);
    System.out.println("Search time in seconds" + timeInSeconds);
    return false;
  }

  // get a list of callables that each search a certain chunk of the array
  private List<Callable<Boolean>> getCallablesForEachChunk(
      int chunkArraySize,
      int numberOfChunksNeeded,
      int valueToFind,
      List<CustomNumberEntity> arrayToSearch) {
    List<Callable<Boolean>> callableList = new ArrayList<>(numberOfChunksNeeded);
    for (int i = 0; i < numberOfChunksNeeded; i++) {
      int startPosForCallable = 0;
      if (i > 0) {
        startPosForCallable = i * chunkArraySize;
      }
      // dont let end pos go out of bounds
      // if the chunk extends past the size, just set endPos as the end of the array
      int endPosForCallable =
          Math.min(startPosForCallable + chunkArraySize - 1, arrayToSearch.size());
      Callable<Boolean> callableSearch =
          new NumberFinderCallable(
              arrayToSearch, valueToFind, startPosForCallable, endPosForCallable);
      callableList.add(callableSearch);
    }
    return callableList;
  }

进行搜索的我的可调用对象

    public class NumberFinderCallable implements Callable<Boolean> {

  private List<CustomNumberEntity> arrayToSearch;
  private int startPos;
  private int endPos;
  private int valueToSearchFor;

  public NumberFinderCallable(
      List<CustomNumberEntity> arrayToSearch, int valueToSearchFor, int startPos, int endPos) {
    this.arrayToSearch = arrayToSearch;
    this.startPos = startPos;
    this.endPos = endPos;
    this.valueToSearchFor = valueToSearchFor;
  }

  @Override
  public Boolean call() {
    System.out.println(
        "Callable started, searching the chunk of array with start pos "
            + startPos
            + " and end pos "
            + endPos);
    for (int i = startPos; i <= endPos; i++) {
      System.out.println(
          "Callable is comparing a number in pos "
              + i
              + " in the chunk with star pos "
              + +startPos
              + " and end pos "
              + endPos);
      if (FastestComparator.compare(valueToSearchFor, arrayToSearch.get(i)) == 0) {
        System.out.println("element found in pos " + i + ". Returning true");
        return true;
      }
    }
    return false;
  }
}

即使在找到真实结果并且所有期货都被取消后，我也可以从日志中看到线程仍在进行中

Answer 1

根据您的previous question，您已经明确表示，您尝试并行化此contains 方法是出于性能原因，而不是为了了解API。但是，IMO 你犯了错误，假设这个方法实际上需要优化。

我对一个包含100000000（1 亿）个元素的列表进行了临时测试，并且对于最坏的情况即试图找到一个列表中不存在的元素。对于顺序contain 方法

list.contains(Integer.MAX_VALUE)

大约平均花费：

0.25 seconds

以及使用流的并行版本：

list.parallelStream().anyMatch(i -> i.equals(Integer.MAX_VALUE))

大约平均花费：

 0.19 seconds

在 4 核机器中加速 1.32x。我非常怀疑一个人会取得比这更多的成就。更不用说以下的可维护性和可读性：

list.parallelStream().anyMatch(i -> i.equals(Integer.MAX_VALUE))

与使用显式执行器等的潜在并行解决方案相比。

如果contain 方法在性能方面非常重要，并且列表中没有重复的元素，则应考虑使用HashSet 来获得恒定复杂度时间contain 方法。

Answer 2

而不是ExecutorService

ExecutorService WORKER_THREAD_POOL = Executors.newFixedThreadPool(NUMBER_THREADS);

使用ThreadPoolExecutor

ThreadPoolExecutor WORKER_THREAD_POOL = (ThreadPoolExecutor) Executors.newFixedThreadPool(NUMBER_THREADS);

然后执行purge() 以释放已取消期货的线程，例如

future.cancel(true);
WORKER_THREAD_POOL.purge();