Golang Concurrent Text Processing

package main

import (
	"bufio"
	"log"
	"os"
)

var concurrency = 100

func main() {
	// This channel has no buffer, so it only accepts input when something is ready
	// to take it out. This keeps the reading from getting ahead of the writers.
	workQueue := make(chan string)

	// We need to know when everyone is done so we can exit.
	complete := make(chan bool)

	// Read the lines into the work queue.
	go func() {
		file, err := os.Open("/path/to/file.csv")
		if err != nil {
			log.Fatal(err)
		}

		// Close when the functin returns
		defer file.Close()

		scanner := bufio.NewScanner(file)

		for scanner.Scan() {
			workQueue <- scanner.Text()
		}

		// Close the channel so everyone reading from it knows we're done.
		close(workQueue)
	}()

	// Now read them all off, concurrently.
	for i := 0; i < concurrency; i++ {
		go startWorking(workQueue, complete)
	}

	// Wait for everyone to finish.
	for i := 0; i < concurrency; i++ {
		<-complete
	}
}

func startWorking(queue chan string, complete chan bool) {
	for line := range queue {
		// Do the work with the line.
	}

	// Let the main process know we're done.
	complete <- true
}

You can use sync.WaitGroup instead completion channel

Hi, I had a question. Wouldn't the concurrent go routines startWorking read the same set of lines since the same workQueue channel is passed in as arg?

Hmm... I don't even remember writing this, haha. But from what I remember of Go, a channel is just a thread-safe queue. So you can have multiple readers and writers without any reader ever seeing a duplicate (unless there's a way to "peek", which I don't think there is).

Thanks for your reply. I am pretty new to Go so I wasn't sure how would that work. After making a few tests, I can confirm the go routines use shared channel as a thread-safe queue.