多线程 - 计算来自多个文件的总字数

Question

我做了一个程序来计算单个文件中的单词， 但是我怎样才能修改我的程序，所以它给出了所有文件中的总字数（作为一个值）。

我的代码如下所示：

public class WordCount implements Runnable
{
   public WordCount(String filename)
   {
      this.filename = filename;
   }

   public void run()
   {
      int count = 0;
      try
      {
         Scanner in = new Scanner(new File(filename));

         while (in.hasNext())
         {
            in.next();
            count++;
         }
         System.out.println(filename + ": " + count);
      }
      catch (FileNotFoundException e)
      {
         System.out.println(filename + " blev ikke fundet.");
      }
   }
   private String filename;
}

使用主类：

public class Main
{

   public static void main(String args[])
   {
      for (String filename : args)
      {
         Runnable tester = new WordCount(filename);

         Thread t = new Thread(tester);
         t.start();
      }
   }
}

以及如何避免竞争条件？ 感谢您的帮助。

Answer 1

一个工作线程：

class WordCount extends Thread
{

   int count;

   @Override
   public void run()
   {
      count = 0;
      /* Count the words... */
      ...
      ++count;
      ...
   }

}

还有一个使用它们的类：

class Main
{

   public static void main(String args[]) throws InterruptedException
   {
      WordCount[] counters = new WordCount[args.length];
      for (int idx = 0; idx < args.length; ++idx) {
         counters[idx] = new WordCount(args[idx]);
         counters[idx].start();
      }
      int total = 0;
      for (WordCount counter : counters) {
        counter.join();
        total += counter.count;
      }
      System.out.println("Total: " + total);
   }

}

许多硬盘不能很好地同时读取多个文件。参考位置对性能有很大影响。

Answer 2

您可以使用Future 来获取计数，最后将所有计数相加或使用静态变量并以synchronized 方式递增它，即显式使用synchronized 或使用Atomic Increment

Answer 3

如果你的Runnable 有两个参数会怎样：

• BlockingQueue<String> 或 BlockingQueue<File> 的输入文件
• AtomicLong

在一个循环中，您将从队列中获取下一个字符串/文件，计算其字数，并将AtomicLong 增加该数量。循环是while(!queue.isEmpty()) 还是while(!done) 取决于您如何将文件送入队列：如果您从一开始就知道所有文件，则可以使用isEmpty 版本，但如果您从某个地方将它们流式传输，您想使用 !done 版本（并让 done 成为 volatile boolean 或 AtomicBoolean 以实现内存可见性）。

然后你将这些Runnables 提供给执行者，你应该很高兴。

Answer 4

您可以创建一些侦听器以从线程中获取反馈。

   public interface ResultListener {
       public synchronized void result(int words);
   }
   private String filename;
   private ResultListener listener;
   public void run()
   {
     int count = 0;
     try
     {
       Scanner in = new Scanner(new File(filename));

       while (in.hasNext())
       {
          in.next();
          count++;
       }
       listener.result(count); 
    }
    catch (FileNotFoundException e)
    {
       System.out.println(filename + " blev ikke fundet.");
    }
   }
  }

您可以为侦听器添加一个构造函数参数，就像为您的文件名一样。

  public class Main
  {
     private static int totalCount = 0;
     private static ResultListener listener = new ResultListener(){
         public synchronized void result(int words){
            totalCount += words;
         }
     }
     public static void main(String args[])
     {
        for (String filename : args)
        {
           Runnable tester = new WordCount(filename, listener);

           Thread t = new Thread(tester);
           t.start();
        }
     }
  }

Answer 5

您可以创建count volatile 和static，以便所有线程都可以递增它。

public class WordCount implements Runnable
{
   private static AtomicInteger count = new AtomicInteger(0); // <-- now all threads increment the same count

   private String filename;

   public WordCount(String filename)
   {
      this.filename = filename;
   }

   public static int getCount()
   {
       return count.get();
   }

   public void run()
   {
      try
      {
         Scanner in = new Scanner(new File(filename));

         while (in.hasNext())
         {
            in.next();
            count.incrementAndGet();
         }
         System.out.println(filename + ": " + count);
      }
      catch (FileNotFoundException e)
      {
         System.out.println(filename + " blev ikke fundet.");
      }
   }
}

更新：已经有一段时间没有做 java 了，但关于将其设为私有静态字段的观点仍然存在……只需将其设为 AtomicInteger。

Answer 6

您可以创建一个带有同步任务队列的线程池，该队列将保存您希望计算字数的所有文件。

当您的线程池工作人员上线时，他们可以向任务队列询问要计数的文件。 在worker完成他们的工作之后，他们可以通知主线程他们的最终编号。

主线程会有一个同步的通知方法，它将所有工作线程的结果相加。

希望这会有所帮助。

Answer 7

或者您可以让所有线程更新单个字数变量。如果 count 是单字的，count++ 是原子的（一个 int 就足够了）。

编辑：事实证明，Java 规范已经够傻了，count++ 是 不是 原子的。我不知道为什么。无论如何，看看 AtomicInteger 和它的 incrementAndGet 方法。希望这个 是原子的（我现在不知道会发生什么......），并且您不需要任何其他同步机制 - 只需将您的计数存储在 AtomicInteger 中。

Answer 8

给定的解决方案是考虑到 Java8 并发包，其中涉及 Executors 和 Future 用于多线程。

首先，为处理单个文件而创建的可调用类

public class WordCounter implements Callable {

Path bookPath;

public WordCounter(Path bookPath) {
    this.bookPath = bookPath;
}

@Override
public Map<String, Long> call() throws Exception {
    Map<String, Long> wordCount = new HashMap<>();
    wordCount = Files.lines(bookPath).flatMap(line -> Arrays.stream(line.trim().split(" ")).parallel())
               .map(word -> word.replaceAll("[^a-zA-Z]", "").toLowerCase().trim())
               .filter(word -> word.length() > 0)
               .map(word -> new SimpleEntry<>(word, 1))
               .collect(Collectors.groupingBy(SimpleEntry::getKey, Collectors.counting()));     
    return wordCount;
}
}

现在，我们将创建多个未来任务来调用/处理参数中的每个文件，如下所示

ExecutorService exes = Executors.newCachedThreadPool();
FutureTask[] tasks = new FutureTask[count];
Map<String, Long> result = new HashMap<>();

Path[] books = new Path[2];
books[0] = Paths.get("C:\\Users\\Documents\\book1.txt");
books[1] = Paths.get("C:\\Users\\Documents\\book2.txt");
    
   for(int i=0; i<books.length; i++) {
        tasks[i] = new FutureTask(new WordCounter(books[i]));
        exes.submit(tasks[i]);
    }
    
    for(int i=0; i<count; i++) {
        try {
            Map<String, Long> wordCount = (Map<String, Long>) tasks[i].get();
            wordCount.forEach((k,v) -> result.put(k, result.getOrDefault(k, 0L)+1));
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
    }
    exes.shutdown();

进一步的result 映射可以升级为 volatile 关键字并在WordCounter 线程之间共享以同时更新字数。

最终结果：result.size() 应该会给出预期的输出