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

Question

我正在使用 Golang，我试图让这个程序线程安全。它以一个数字作为参数（即要启动的消费者任务的数量），从输入中读取行，并累积字数。我希望线程是安全的（但我不希望它只是锁定所有内容，它需要高效）我应该使用通道吗？我该怎么做呢？

package main

import (

    "bufio"

    "fmt"

    "log"

    "os"

    "sync"

)

// Consumer task to operate on queue

func 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 queue

var 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) {&nbsp; &nbsp; fmt.Printf("I'm consumer task #%v\n", task_num)&nbsp; &nbsp; for item := range ch {&nbsp; &nbsp; &nbsp; &nbsp; fmt.Printf("task %d consuming: Line item: %v\n", task_num, item)&nbsp; &nbsp; }&nbsp; &nbsp; // each worker will drop out of their loop when channel is closed}从切片更改queue为频道和 feed 项目，如下所示：queue := make(chan string)go func() {&nbsp; &nbsp; // Loop through each line in the file and append it to the queue&nbsp; &nbsp; for scanner.Scan() {&nbsp; &nbsp; &nbsp; &nbsp; queue <- scanner.Text()&nbsp; &nbsp; }&nbsp; &nbsp; close(queue) // signal to workers that there is no more items}()然后只需更新您的工作调度程序代码以添加频道输入：go func(i int) {&nbsp; &nbsp; consumer_task(i, queue) // add the queue parameter&nbsp; &nbsp; wg.Done()}(i)https://go.dev/play/p/AzHyztipUZI