ทิศทางแชนเนลและรูปแบบไปป์ไลน์

ชนิดช่องสัญญาณแบบกำหนดทิศทาง

package main

// send-only: can only send, not receive or close (close is allowed by sender)
func sender(out chan<- int) {
    out <- 42
}

// receive-only: can only receive, not send or close
func receiver(in <-chan int) int {
    return <-in
}

// bidirectional: can do both
func relay(in <-chan int, out chan<- int) {
    out <- <-in
}

การแปลงทิศทางของช่องสัญญาณ

package main
import "fmt"

func produce(out chan<- int) { out <- 1; close(out) }
func consume(in <-chan int)  { fmt.Println(<-in) }

func main() {
    ch := make(chan int)   // bidirectional
    go produce(ch)         // auto-converts to chan<- int
    consume(ch)            // auto-converts to <-chan int
    // chan<- int cannot be converted back to <-chan int
}

ไปป์ไลน์: ขั้นที่ 1 — ตัวสร้าง

package main
import "fmt"

func generate(nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        for _, n := range nums {
            out <- n
        }
        close(out)
    }()
    return out
}

func main() {
    ch := generate(2, 3, 4, 5)
    for v := range ch { fmt.Println(v) }
}

ไปป์ไลน์: ขั้นที่ 2 — การแปลง

package main
import "fmt"

func generate(nums ...int) <-chan int {
    out := make(chan int)
    go func() { for _, n := range nums { out <- n }; close(out) }()
    return out
}

func square(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        for v := range in { out <- v * v }
        close(out)
    }()
    return out
}

func main() {
    for v := range square(generate(2, 3, 4, 5)) {
        fmt.Println(v)  // 4, 9, 16, 25
    }
}

ไปป์ไลน์: ตัวอย่างแบบสามขั้นตอนครบถ้วน

package main
import "fmt"

func gen(nums ...int) <-chan int {
    c := make(chan int)
    go func() { for _, n := range nums { c <- n }; close(c) }()
    return c
}

func double(in <-chan int) <-chan int {
    c := make(chan int)
    go func() { for v := range in { c <- v * 2 }; close(c) }()
    return c
}

func evens(in <-chan int) <-chan int {
    c := make(chan int)
    go func() { for v := range in { if v%2 == 0 { c <- v } }; close(c) }()
    return c
}

func main() {
    for v := range evens(double(gen(1,2,3,4,5))) {
        fmt.Println(v)  // 2, 4, 6, 8, 10
    }
}

การยกเลิกไปป์ไลน์ด้วยช่องสัญญาณ done

package main
import "fmt"

func gen(done <-chan struct{}, nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for _, n := range nums {
            select {
            case out <- n:
            case <-done:  // cancel signal
                return
            }
        }
    }()
    return out
}

func main() {
    done := make(chan struct{})
    ch := gen(done, 1, 2, 3, 4, 5)
    fmt.Println(<-ch)  // 1
    close(done)        // cancel remaining work
    fmt.Println("pipeline cancelled")
}

การกระจายงาน: แบ่งงาน

package main
import ("fmt"; "sync")

func fanOut(in <-chan int, workers int) []<-chan int {
    outs := make([]<-chan int, workers)
    for i := 0; i < workers; i++ {
        out := make(chan int)
        outs[i] = out
        go func() {
            for v := range in { out <- v * v }  // each worker processes
            close(out)
        }()
    }
    return outs
}

func main() {
    fmt.Println("fan-out distributes work to multiple goroutines")
    _ = sync.WaitGroup{}
}

การรวมช่องสัญญาณ: รวมช่องสัญญาณ

package main
import ("fmt"; "sync")

func merge(cs ...<-chan int) <-chan int {
    out := make(chan int)
    var wg sync.WaitGroup
    for _, c := range cs {
        wg.Add(1)
        c := c
        go func() { defer wg.Done(); for v := range c { out <- v } }()
    }
    go func() { wg.Wait(); close(out) }()
    return out
}

func gen(nums ...int) <-chan int {
    c := make(chan int)
    go func() { for _, n := range nums { c <- n }; close(c) }()
    return c
}

func main() {
    for v := range merge(gen(1,2), gen(3,4), gen(5,6)) {
        fmt.Println(v)
    }
}

แนวทางปฏิบัติที่ดีสำหรับไปป์ไลน์

ไปป์ไลน์ในโลกจริง: การประมวลผล CSV

package main
import ("fmt"; "strings")

func splitLines(data string) <-chan string {
    out := make(chan string)
    go func() {
        for _, line := range strings.Split(data, "\n") {
            if line != "" { out <- line }
        }
        close(out)
    }()
    return out
}

func main() {
    csv := "alice,30\nbob,25\ncarol,35"
    for line := range splitLines(csv) {
        parts := strings.Split(line, ",")
        fmt.Printf("name=%s age=%s\n", parts[0], parts[1])
    }
}

ตรวจสอบความเข้าใจอย่างรวดเร็ว

ทบทวน: ทิศทางของช่องสัญญาณและไปป์ไลน์