请帮我看看 go 管道的实现(还在学习这门语言)

大家好，

我正在学习 go 和它的并发功能，“管道”是常见模式之一。下面是我的实现，想请大家看看，并建议还有什么可以改进的地方

package main

import (
    "context"
    "fmt"
    "runtime"
    "sync"
    "time"
)

func generate(ctx context.Context, inputs chan<- int) {
    defer close(inputs)
    item := 0
    for {
        select {
        case inputs ← item:
            item++
            time.Sleep(60 * time.Millisecond)
        case <-ctx.Done():
            fmt.Println("Ending producer goroutine and inputs channel gracefuly.")
            return
        }
    }
}

func process(inputs <-chan int, outputs chan<- int, group *sync.WaitGroup) {
    defer group.Done()
    for item := range inputs {
        outputs ← item * item
    }
}

func collect(outputs <-chan int) {
    for square := range outputs {
        fmt.Printf("square %d\n", square)
    }
}

func finish(group *sync.WaitGroup, outputs chan<- int) {
    group.Wait()
    close(outputs)
    fmt.Println("Ending coordinator/worker goroutines and outputs channel gracefuly.")
}

func main() {
    ctx, cancel := context.WithTimeout(
        context.Background(),
        1800*time.Millisecond
    )
    defer cancel()

    inputs := make(chan int, 960)
    outputs := make(chan int, 960)

    go generate(ctx, inputs)

    var group sync.WaitGroup
    for range 4 {
        group.Add(1)
        go process(inputs, outputs, &group)
    }

    go finish(&group, outputs)

    collect(outputs)

    fmt.Printf("Pipeline drained completly !\n")
    fmt.Printf("Goroutine count : %d\n", runtime.NumGoroutine()) // expected to be 1
}

另外，还有哪些重要且常用的并发模式值得学习，比如 pubsub。

祝好，

FV