为什么第二个通道没有从前一个通道接收数据

Question

我正在使用 go 通道执行我的第一步，并尝试编写以下代码，我想在其中使用从给定 csv 文件读取的 goroutine 执行以下操作，将读取记录发送到正在添加的另一个通道相同的记录到另一个 csv 文件：

package main

import (
    "encoding/csv"
    "encoding/json"
    "fmt"
    "log"
    "os"
)

func failOnError(err error) {
    if err != nil {
        log.Fatal("Error:", err)
        panic(err)
    }
}
func main() {
    read := make(chan Data)
    go func(input_file string) {
        var data Data

        fr, err := os.Open(input_file)
        failOnError(err)
        defer fr.Close()
        r := csv.NewReader(fr)
        rows, err := r.ReadAll()
        failOnError(err)
        data.header = rows[0]
        for _, row := range rows[1:] {
            data.lines = append(data.lines, Person{
                Firstname: row[0],
                Lastname:  row[1],
                Address: &Address{
                    City:  row[2],
                    State: row[3],
                },
            })
        }

        peopleJson, _ := json.Marshal(data.lines)
        fmt.Println(string(peopleJson))   // This is working smoothly

        read <- data
    }("people.csv")

    csvOut, err := os.Create("resultsfile.csv")
    if err != nil {
        log.Fatal("Unable to open output")
    }

    out := make(chan int)
    select {
    case data := <-read:
        go func(data Data) {
            println("data received")     // <-- Not show up
            w := csv.NewWriter(csvOut)
            defer csvOut.Close()
            // handle header
            data.header = append(data.header, "score")
            if err = w.Write(data.header); err != nil {
                log.Fatal(err)
            }
            /*
                hanlde data
            */
            w.Flush()
            out <- 0
        }(data)
    case _ = <-out:
        println("done")
    }

}

type Person struct {
    Firstname string   `json:"firstname"` // JSON annotation will allow for easy printing to JSON after it had been loaded
    Lastname  string   `json:"lastname"`
    Address   *Address `json:"address,omitempty"`
}

type Address struct {
    City  string `json:"city"`
    State string `json:"state"`
}

type Data struct {
    header []string
    lines  []Person
}

我的代码失败，没有错误并且没有显示data received 注释

people.csv 文件是：

First,Last,City,State
Nic,Raboy,San Francisco,CA
Maria,Raboy,Dublin,CA
Hasan,yousef, Jordan, ZA

我得到的输出是：

[Running] go run "d:\goplay\csv.go"
[{"firstname":"Nic","lastname":"Raboy","address":{"city":"San Francisco","state":"CA"}},{"firstname":"Maria","lastname":"Raboy","address":{"city":"Dublin","state":"CA"}},{"firstname":"Hasan","lastname":"yousef","address":{"city":" Jordan","state":" ZA"}}]

[Done] exited with code=0 in 1.233 seconds

Answer 1

问题是这样的：

    select {
    case data := <-read:
        go func(data Data) {

select 是 main 函数中的最后一件事。当它从read 读取data 时，它会在后台启动一个goroutine 来处理它。然后main 函数结束，因此程序退出。

Go 不会自动等待 goroutines。您可以使用“完成”通道或其他任何方式在它们之间进行同步。

你基本上有一个管道，在概念上看起来像这样：

                                         ----> process -> write to `out`
                                        /
start --> main file waits  data on `read` *or* wait for `out` -> done
       \
        ---> CSV reads all lines and outputs data to `read`  

因为您实际上只通过频道发送一条消息，所以您实际上并不需要select 或loop。您只需要等待数据，处理它，然后让main 等待完成：

package main

import (
    "encoding/csv"
    "encoding/json"
    "fmt"
    "log"
    "os"
)

func failOnError(err error) {
    if err != nil {
        log.Fatal("Error:", err)
        panic(err)
    }
}
func main() {
    read := make(chan Data)
    go func(input_file string) {
        var data Data

        fr, err := os.Open(input_file)
        failOnError(err)
        defer fr.Close()
        r := csv.NewReader(fr)
        rows, err := r.ReadAll()
        failOnError(err)
        data.header = rows[0]
        for _, row := range rows[1:] {
            data.lines = append(data.lines, Person{
                Firstname: row[0],
                Lastname:  row[1],
                Address: &Address{
                    City:  row[2],
                    State: row[3],
                },
            })
        }

        peopleJson, _ := json.Marshal(data.lines)
        fmt.Println(string(peopleJson)) // This is working smoothly

        read <- data
    }("people.csv")

    csvOut, err := os.Create("resultsfile.csv")
    if err != nil {
        log.Fatal("Unable to open output")
    }

    out := make(chan int)
    go func() {
        data := <-read
        println("data received") // <-- Not show up
        w := csv.NewWriter(csvOut)
        defer csvOut.Close()
        // handle header
        data.header = append(data.header, "score")
        if err = w.Write(data.header); err != nil {
            log.Fatal(err)
        }
        /*
           hanlde data
        */
        w.Flush()
        out <- 0
    }()

    <-out
    println("done")
}

type Person struct {
    Firstname string   `json:"firstname"` // JSON annotation will allow for easy printing to JSON after it had been loaded
    Lastname  string   `json:"lastname"`
    Address   *Address `json:"address,omitempty"`
}

type Address struct {
    City  string `json:"city"`
    State string `json:"state"`
}

type Data struct {
    header []string
    lines  []Person
}

OP 询问，

有没有办法在没有选择的情况下从以前的频道接收输入，同时确保程序在完成所需内容之前不退出？ – Hasan A Yousef 1 分钟前

您所写的所有频道都只传递一条消息，这意味着一旦您阅读了该频道一次，您就可以认为作者是完整的。因此，通过使用out 频道，您已经实现了这一点。

如果您有 未知 数量的发送到通道，您可以使用内置的 close 功能来实现此目的：

package main

import (
  "fmt"
)

func main() {
  send := make(chan int)
  result := make(chan int)
  go func() {
    sum := 0
    for i := range send {
      sum += i
    }
    result <- sum
  }()
  for _, x := range []int{1,2,3}  {
    send <- x
  }
  close(send)
  fmt.Println("Sum is: ", <-result)
}