bufio包系列之一个误用bufio读取的示例

var linesPool = sync.Pool{
  New: func() interface{} {
    lines := make([]byte, 250*1024)
    return lines
  },
}

var stringPool = sync.Pool{
  New: func() interface{} {
    lines := ""
    return lines
  },
}

func main() {
  s := time.Now()
  flag.Parse()
  if *filename == "" {
    panic("please input a filename")
  }
  file, err := os.Open(*filename)

  if err != nil {
    fmt.Println("cannot able to read the file", err)
    return
  }
  defer file.Close() //close after checking err

  Process(file)
  fmt.Println("\nTime taken - ", time.Since(s))
}

func Process(f *os.File) error {
  r := bufio.NewReader(f)

  var wg sync.WaitGroup
  for {
    buf := linesPool.Get().([]byte)
    n, err := r.Read(buf)
    buf = buf[:n]

    if n == 0 {
      // 这里做了简化，直接跳出循环
      break
    }
    // 为了确保buf中都是整行整行的读取
    nextUntillNewline, err := r.ReadBytes('\n')

    if err != io.EOF {
      buf = append(buf, nextUntillNewline...)
    }

    wg.Add(1)
    go func() {
      ProcessChunk(buf)
      wg.Done()
    }()
  }

  wg.Wait()
  return nil
}

func ProcessChunk(chunk []byte) {
  var wg2 sync.WaitGroup

  logs := stringPool.Get().(string)
  logs = string(chunk)
  linesPool.Put(chunk)

  logsSlice := strings.Split(logs, "\n")

  stringPool.Put(logs)

  // 每300行启用一个协程进行处理
  chunkSize := 300
  // 读取到的行数
  lines := len(logsSlice)
  // 总共需要的协程数
  numberOfThreads := lines / chunkSize

  //如果还有余数，则多加1个协程
  if lines%chunkSize != 0 {
    numberOfThreads++
  }

  for i := 0; i < (numberOfThreads); i++ {
    wg2.Add(1)
    // startLine是从第几行开始处理
    // endLine是结束的行
    startLine := i * chunkSize
    endLine := int(math.Min(float64((i+1)*chunkSize), float64(len(logsSlice))))
    go func(startLine int, endLine int) {
      defer wg2.Done() //to avaoid deadlocks
      for i := startLine; i < endLine; i++ {
        text := logsSlice[i]
        if len(text) == 0 {
          continue
        }
        //进行文本的处理分析
        fmt.Println(text)
      }
    }(startLine, endLine)
  }

  wg2.Wait()
  logsSlice = nil
}