时间戳

首页 > 解决方案 > 多线程（ExecutorService）方法意外工作

java - 多线程（ExecutorService）方法意外工作

问题描述

我无法理解错误在哪里。我正在尝试制作一个多线程程序，该程序将在数组中找到那些具有相同符号的字符串。然后这些字符串应该与它们的索引一起打印。

Example input: {"qwe","wqe","qwee","a","a"};
Output:
a = 3, 4
eqw = 0, 1

但是当我试图运行程序时，它们的索引发生了一些事情。我尝试以这样的方式同步我的方法：使用（this）同步，在方法级别同步，使用为每个对象创建的锁同步。也许我错过了一些东西，并且有一些东西可以让它发挥作用？

 public class Main {

  private volatile Map<String, Integer> countByString = new ConcurrentHashMap<>();
  private volatile Map<String, String> indexesByString = new ConcurrentHashMap<>();

  public static void main(String[] args) {
    String[] arr = {"qwe", "wqe", "qwee", "a", "a"};
    Main main = new Main();
    List<Callable<Void>> tasks = new ArrayList<>();
    AtomicInteger i = new AtomicInteger(0);
    Arrays.stream(arr).forEach(str -> {
      tasks.add(() -> {
        char[] charArr = str.toCharArray();
        Arrays.sort(charArr);
        String sortedStr = new String(charArr);
        main.calculateCount(sortedStr);
        main.calculateIndex(sortedStr, i.getAndIncrement());
        return null;
      });
    });

    try {
      ExecutorService executorService = Executors.newFixedThreadPool(4);
      executorService.invokeAll(tasks);
      executorService.shutdown();
    } catch (Exception e) {
      e.printStackTrace();
    }
    main.printResult();
  }

  void calculateCount(String str) {
    synchronized (countByString) {
      int count = countByString.get(str) == null ? 0 : countByString.get(str);
      countByString.put(str, ++count);
    }
  }

  void calculateIndex(String str, int index) {
    synchronized (indexesByString) {
      System.out.println(Thread.currentThread().getName() + " " + str + " " + index);
      String indexes = indexesByString.get(str);
      if (indexes == null) {
        indexes = "";
      }
      indexes += (index + ";");
      indexesByString.put(str, indexes);
    }
  }

  private void printResult() {
    for (Map.Entry<String, Integer> entry : countByString.entrySet()) {
      String str = entry.getKey();
//      Integer count = entry.getValue();
//      if (count >= 2) {
      String indexes = indexesByString.get(str);
      System.out.println(str + " = " + indexes);
//      }
    }
  }
}

标签： javamultithreadingparallel-processingthread-safetyexecutorservice

解决方案

您的代码的问题是这部分：

Arrays.stream(arr).forEach(str -> {
  tasks.add(() -> {
    char[] charArr = str.toCharArray();
    Arrays.sort(charArr);
    String sortedStr = new String(charArr);
    main.calculateCount(sortedStr);
    main.calculateIndex(sortedStr, i.getAndIncrement());
    return null;
  });
});

您正在将相同的工作（在整个字符串数组中搜索）分配给所有线程。因此，线程会覆盖彼此的索引。相反，您应该做的是在线程之间分配搜索字符串的工作。类似于：

tasks.add(() -> {
for(int threadIndex = i.getAndIncrement(); threadIndex < arr.length; threadIndex = i.getAndIncrement()){
        char[] charArr = arr[threadIndex].toCharArray();
        Arrays.sort(charArr);
        String sortedStr = new String(charArr);
        main.calculateCount(sortedStr);
        main.calculateIndex(sortedStr, threadIndex);
}
return null;
});

我使用 的线程安全属性AtomicInteger来模拟线程之间循环迭代的动态工作分配。

您提供的代码还有其他问题，主要与性能和同步开销有关。

private volatile Map<String, Integer> countByString = new ConcurrentHashMap<>();
private volatile Map<String, String> indexesByString = new ConcurrentHashMap<>();

在这种情况下，您既不需要volatile也不需要ConcurrentHashMap，但是，您可以创建这些字段final。由于您在并行区域之前volatile创建对象，因此可以删除该子句，并且可以将其更改为 a ，因为您已经在同步这些字段。mainConcurrentHashMapHashMap

在关注代码的正确性之后，您应该尽量减少同步开销。例如，您可以尝试在每个线程之间复制countByString和indexesByString，并在并行工作完成后依次减少它们的值。

自然，鉴于您当前输入的大小，很难注意到性能优化之间的有意义的差异。

推荐阅读