使用渠道进行管道

Question

I'm exploring Go and trying to set up a sort of pipeline using channels. I just want to read something in main() and send them to process() for processing, in this case just print the value to the screen.

Unfortunately, in the code below, it appears that process() never reads from the channel, or at least it doesn't print anything; what am I doing wrong?

package main

import ( "fmt" ; "database/sql" ; _ "github.com/lib/pq" ; "time" ; "gopkg.in/redis.v3" )//; "strconv" )

type Record struct {
    userId, myDate int
    prodUrl string
}


func main(){

    //connect to db
    db, err := sql.Open(...)
    defer db.Close()

    //error check here...

    //exec query
    rows, err := db.Query("select userID,url,date from mytable limit 10")
    defer rows.Close()

    //error check here...   

    //create channel to buffer rows read
    bufferChan := make(chan *Record,1000)
    go process(bufferChan)

    //iterate through results and send them to process()
    row := new(Record)
    for rows.Next(){
        err := rows.Scan(&row.userId, &row.prodUrl, &row.myDate)        
        bufferChan <- row
        fmt.Printf("row sent %v",row.userId)                    
    }   
}

//prints Record values
func process (buffer chan *Record) {
    row := <- buffer
    fmt.Printf("row received: %d %v %d ", row.userId,row.prodUrl,row.myDate)
}

Answer 1

The reason for func process not printing anything is that you func main exits after the for loop for rows.Next finishes thereby exiting the program. You need to do couple of things.

1. Add call to close after for loop to indicate end adding message to buffered channel else it can lead to deadlock. So call close(bufferChan)
2. Use range to iterate over channel in your func process.
3. Pass an additional channel to process to know when it finishes so that main can wait till process finishes.

Look at the code snippet below for example:

package main

import "fmt"

func main() {
    bufferChan := make(chan int, 1000)
    done := make(chan bool)
    go process(bufferChan, done)
    for i := 0; i < 100; i++ {
        bufferChan <- i
    }
    close(bufferChan)

    select {
    case <-done:
        fmt.Println("Done")
    }

}

func process(c chan int, done chan bool) {
    for s := range c {
        fmt.Println(s)
    }   
    done <- true

}