我如何使这个程序线程安全，通道是否是最好的实现，如果是，如何实现？

Question

我正在使用 Golang，我试图让这个程序线程安全。它以一个数字作为参数（即要启动的消费者任务的数量），从输入中读取行，并累积字数。我希望线程是安全的（但我不希望它只是锁定所有内容，它需要高效）我应该使用通道吗？我该怎么做呢？package mainimport (    "bufio"    "fmt"    "log"    "os"    "sync")// Consumer task to operate on queuefunc consumer_task(task_num int) {        fmt.Printf("I'm consumer task #%v ", task_num)    fmt.Println("Line being popped off queue: " + queue[0])    queue = queue[1:]}// Initialize queuevar queue = make([]string, 0)func main() {    // Initialize wait group     var wg sync.WaitGroup    // Get number of tasks to run from user    var numof_tasks int    fmt.Print("Enter number of tasks to run: ")    fmt.Scan(&numof_tasks)        // Open file    file, err := os.Open("test.txt")    if err != nil {        log.Fatal(err)    }    defer file.Close()    // Scanner to scan the file    scanner := bufio.NewScanner(file)    if err := scanner.Err(); err != nil {        log.Fatal(err)    }    // Loop through each line in the file and append it to the queue    for scanner.Scan() {        line := scanner.Text()          queue = append(queue, line)    }    // Start specified # of consumer tasks    for i := 1; i <= numof_tasks; i++ {        wg.Add(1)        go func(i int) {             consumer_task(i)             wg.Done()        }(i)    }    wg.Wait()    fmt.Println("All done")    fmt.Println(queue)}

Answer 1

您在切片上存在数据竞争queue。并发 goroutines，当通过sync.Mutex锁以受控方式从队列头部弹出元素时。或者使用一个通道来管理工作项的“队列”。

要将您所拥有的转换为使用通道，请更新工作人员以将输入通道作为您的队列 - 并在通道上进行范围，以便每个工作人员可以处理多个任务：

func consumer_task(task_num int, ch <-chan string) {

    fmt.Printf("I'm consumer task #%v\n", task_num)

    for item := range ch {

        fmt.Printf("task %d consuming: Line item: %v\n", task_num, item)

    }

    // each worker will drop out of their loop when channel is closed

}

从切片更改queue为频道和 feed 项目，如下所示：

queue := make(chan string)

go func() {

    // Loop through each line in the file and append it to the queue

    for scanner.Scan() {

        queue <- scanner.Text()

    }

    close(queue) // signal to workers that there is no more items

}()

然后只需更新您的工作调度程序代码以添加频道输入：

go func(i int) {

    consumer_task(i, queue) // add the queue parameter

    wg.Done()

}(i)

https://go.dev/play/p/AzHyztipUZI