Чтение файла одновременно в Голанге

Считывающая часть не является параллельной, но обрабатывается. Я сформулировал название таким образом, потому что я, скорее всего, снова буду искать эту проблему, используя эту фразу.:)

Я собираюсь зайти в тупик, пытаясь выйти за рамки примеров, поэтому для меня это опыт обучения. Мои цели таковы:

Прочитайте файл по строкам (в конечном итоге используйте буфер для создания групп строк).
Отпустите текст в func(), который выполняет некоторое регулярное выражение.
Присылайте результаты где-нибудь, но избегайте мьютексов или общих переменных. Я посылаю ints (всегда число 1) в канал. Это вроде глупо, но если это не вызывает проблем, я бы хотел оставить это так, если у вас нет опциона.
Используйте рабочий пул для этого. Я не уверен, как я говорю рабочим, чтобы они сами требовали?

Вот ссылка playground. Я попытался написать полезные комментарии, надеюсь, это имеет смысл. Мой дизайн может быть совершенно неправильным, поэтому не стесняйтесь рефакторировать.

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)
}

Ответ 1

Вы почти там, просто нужно немного поработать над синхронизацией goroutines. Ваша проблема в том, что вы пытаетесь передать парсер и собирать результаты в одной и той же процедуре, но это невозможно.

Я предлагаю следующее:

Запустите сканер в отдельной процедуре, закройте входной канал, когда все прочитано.
Запустите отдельную процедуру, ожидающую, когда парсеры закончат свое задание, чем закрыть выходной канал.
Соберите все результаты в вашей основной процедуре.

Соответствующие изменения могут выглядеть так:

// 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
}

Полностью рабочий пример на игровой площадке: http://play.golang.org/p/coja1_w-fY

Стоит добавить, что вам не обязательно нужен WaitGroup для достижения того же самого, все, что вам нужно знать, - это когда прекратить получать результаты. Это может быть достигнуто, например, путем рекламы сканера (по каналу), сколько строк было прочитано, а затем коллектор, считывающий только указанное количество результатов (вам также необходимо отправить нули).

Ответ 2

Изменить: Ответ от @tomasz выше правильный. Пожалуйста, проигнорируйте этот ответ.

Вам нужно сделать две вещи:

используйте буферный chan, чтобы отправка не блокировала
закрыть chan, чтобы получатель не блокировал.

Использование буферизованных каналов имеет важное значение, поскольку для небуферизованных каналов требуется получение для каждой отправки, что вызывает тупик, который вы нажимаете.

Если вы исправите это, вы столкнетесь с тупиком при попытке получить результаты, потому что результаты не были закрыты.

Здесь фиксированная площадка: http://play.golang.org/p/DtS8Matgi5