并发读取文件

Question

读取部分不是并发的，但是处理是并发的。我用这种方式来表达标题，因为我很可能会用这个短语再次搜索这个问题。:)

在尝试超越示例之后，我陷入了僵局，所以这对我来说是一次学习体验。我的目标是：

1. 逐行读取文件(最终使用缓冲区完成行组)。
2. 将文本传递给执行某些正则化工作的func()。
3. 将结果发送到某个地方，但避免互斥或共享变量。我把ints (总是数字1)发送到一个频道。这有点傻，但如果它没有引起问题，我想让它这样，除非你们有一个更整洁的选择。
4. 使用员工池进行此操作。我不知道我怎么叫工人们报答自己？

这是操场链接。我试着写一些有用的评论，希望这是有意义的。我的设计可能是完全错误的，所以不要犹豫重构。

package main

import (
  "bufio"
  "fmt"
  "regexp"
  "strings"
  "sync"
)

func telephoneNumbersInFile(path string) int {
  file := strings.NewReader(path)

  var telephone = regexp.MustCompile(`\(\d+\)\s\d+-\d+`)

  // do I need buffered channels here?
  jobs := make(chan string)
  results := make(chan int)

  // I think we need a wait group, not sure.
  wg := new(sync.WaitGroup)

  // start up some workers that will block and wait?
  for w := 1; w <= 3; w++ {
    wg.Add(1)
    go matchTelephoneNumbers(jobs, results, wg, telephone)
  }

  // go over a file line by line and queue up a ton of work
  scanner := bufio.NewScanner(file)
  for scanner.Scan() {
    // Later I want to create a buffer of lines, not just line-by-line here ...
    jobs <- scanner.Text()
  }

  close(jobs)
  wg.Wait()

  // Add up the results from the results channel.
  // The rest of this isn't even working so ignore for now.
  counts := 0
  // for v := range results {
  //   counts += v
  // }

  return counts
}

func matchTelephoneNumbers(jobs <-chan string, results chan<- int, wg *sync.WaitGroup, telephone *regexp.Regexp) {
  // Decreasing internal counter for wait-group as soon as goroutine finishes
  defer wg.Done()

  // eventually I want to have a []string channel to work on a chunk of lines not just one line of text
  for j := range jobs {
    if telephone.MatchString(j) {
      results <- 1
    }
  }
}

func main() {
  // An artificial input source.  Normally this is a file passed on the command line.
  const input = "Foo\n(555) 123-3456\nBar\nBaz"
  numberOfTelephoneNumbers := telephoneNumbersInFile(input)
  fmt.Println(numberOfTelephoneNumbers)
}

Answer 1

你就快到了，只需要花点功夫来实现哥鲁蒂的同步。您的问题是，您试图在相同的例程中输入解析器并收集结果，但这是无法做到的。

我提出以下建议：

1. 在一个单独的例程中运行扫描器，一旦读取所有内容，就关闭输入通道。
2. 运行单独的例程，等待解析器完成其工作，而不是关闭输出通道。
3. 收集所有的结果，在你的主程序。

相关的修改可以如下所示：

// Go over a file line by line and queue up a ton of work
go func() {
    scanner := bufio.NewScanner(file)
    for scanner.Scan() {
        jobs <- scanner.Text()
    }
    close(jobs)
}()

// Collect all the results...
// First, make sure we close the result channel when everything was processed
go func() {
    wg.Wait()
    close(results)
}()

// Now, add up the results from the results channel until closed
counts := 0
for v := range results {
    counts += v
}

操场上完全工作的示例：w-fY

值得补充的是，您不一定需要WaitGroup来实现同样的目标，您只需要知道什么时候停止接收结果。例如，通过扫描仪广告(在频道上)读取多少行，然后收集器只读取指定数量的结果(您也需要发送零)，就可以实现这一点。

Answer 2

编辑：上面@的答案是正确的。请不要理会这个答案。

你需要做两件事：

1. 使用缓冲的chan，这样发送就不会阻塞
2. 关闭结果chan，以便接收不会阻塞。

使用缓冲通道是非常重要的，因为未缓冲的通道需要对每个发送进行接收，这将导致您所遇到的死锁。

如果您修复了这个问题，您将在尝试接收结果时陷入死锁，因为结果尚未关闭。

这是固定操场：http://play.golang.org/p/DtS8Matgi5