title: "Go 并发模式：从 Goroutine 到 Channel 的实战指南"
date: "2026-07-10"
tags: ["Go", "并发", "Goroutine", "Channel"]
Go 并发模式：从 Goroutine 到 Channel 的实战指南
Go 的并发模型是其最强大的特性之一。理解 goroutine 和 channel 的正确用法，是写出高性能 Go 程序的基础。
Goroutine 基础
GO
package main
import (
"fmt"
"time"
)
func processTask(id int) {
fmt.Printf("任务 %d 开始\n", id)
time.Sleep(time.Second)
fmt.Printf("任务 %d 完成\n", id)
}
func main() {
// 启动 5 个 goroutine
for i := 0; i < 5; i++ {
go processTask(i)
}
// 等待所有 goroutine 完成
time.Sleep(2 * time.Second)
}WaitGroup
GO
package main
import (
"fmt"
"sync"
"time"
)
func worker(id int, wg *sync.WaitGroup) {
defer wg.Done()
fmt.Printf("Worker %d 开始\n", id)
time.Sleep(time.Second)
fmt.Printf("Worker %d 完成\n", id)
}
func main() {
var wg sync.WaitGroup
for i := 0; i < 5; i++ {
wg.Add(1)
go worker(i, &wg)
}
wg.Wait()
fmt.Println("所有 worker 完成")
}Channel 基础
GO
package main
import "fmt"
func producer(ch chan<- int) {
for i := 0; i < 5; i++ {
ch <- i
}
close(ch)
}
func consumer(ch <-chan int) {
for val := range ch {
fmt.Printf("收到: %d\n", val)
}
}
func main() {
ch := make(chan int)
go producer(ch)
consumer(ch)
}工作池模式
GO
package main
import (
"fmt"
"sync"
"time"
)
type Job struct {
ID int
Payload string
}
type Result struct {
JobID int
Output string
}
func worker(id int, jobs <-chan Job, results chan<- Result, wg *sync.WaitGroup) {
defer wg.Done()
for job := range jobs {
// 处理任务
time.Sleep(100 * time.Millisecond)
results <- Result{
JobID: job.ID,
Output: fmt.Sprintf("Worker %d 处理了 %s", id, job.Payload),
}
}
}
func main() {
const numWorkers = 5
const numJobs = 20
jobs := make(chan Job, numJobs)
results := make(chan Result, numJobs)
var wg sync.WaitGroup
// 启动 worker
for i := 0; i < numWorkers; i++ {
wg.Add(1)
go worker(i, jobs, results, &wg)
}
// 发送任务
for i := 0; i < numJobs; i++ {
jobs <- Job{ID: i, Payload: fmt.Sprintf("task-%d", i)}
}
close(jobs)
// 等待所有 worker 完成
wg.Wait()
close(results)
// 收集结果
for result := range results {
fmt.Println(result.Output)
}
}Fan-Out / Fan-In
GO
package main
import (
"fmt"
"sync"
)
func fanOut(input <-chan int, numWorkers int) []<-chan int {
channels := make([]<-chan int, numWorkers)
for i := 0; i < numWorkers; i++ {
channels[i] = process(input)
}
return channels
}
func process(input <-chan int) <-chan int {
output := make(chan int)
go func() {
defer close(output)
for val := range input {
output <- val * 2
}
}()
return output
}
func fanIn(channels ...<-chan int) <-chan int {
var wg sync.WaitGroup
output := make(chan int)
for _, ch := range channels {
wg.Add(1)
go func(c <-chan int) {
defer wg.Done()
for val := range c {
output <- val
}
}(ch)
}
go func() {
wg.Wait()
close(output)
}()
return output
}
func main() {
input := make(chan int)
go func() {
for i := 0; i < 10; i++ {
input <- i
}
close(input)
}()
channels := fanOut(input, 3)
merged := fanIn(channels...)
for val := range merged {
fmt.Println(val)
}
}Context 控制
GO
package main
import (
"context"
"fmt"
"time"
)
func longRunningTask(ctx context.Context) error {
select {
case <-time.After(5 * time.Second):
fmt.Println("任务完成")
return nil
case <-ctx.Done():
fmt.Println("任务取消:", ctx.Err())
return ctx.Err()
}
}
func main() {
// 超时控制
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
defer cancel()
err := longRunningTask(ctx)
if err != nil {
fmt.Println("错误:", err)
}
}并发安全
GO
package main
import (
"fmt"
"sync"
)
type Counter struct {
mu sync.RWMutex
value int
}
func (c *Counter) Increment() {
c.mu.Lock()
defer c.mu.Unlock()
c.value++
}
func (c *Counter) Value() int {
c.mu.RLock()
defer c.mu.RUnlock()
return c.value
}
func main() {
counter := &Counter{}
var wg sync.WaitGroup
for i := 0; i < 1000; i++ {
wg.Add(1)
go func() {
defer wg.Done()
counter.Increment()
}()
}
wg.Wait()
fmt.Println("最终值:", counter.Value()) // 1000
}Pipeline 模式
GO
package main
import "fmt"
func generator(nums ...int) <-chan int {
out := make(chan int)
go func() {
defer close(out)
for _, n := range nums {
out <- n
}
}()
return out
}
func square(in <-chan int) <-chan int {
out := make(chan int)
go func() {
defer close(out)
for n := range in {
out <- n * n
}
}()
return out
}
func filter(in <-chan int, threshold int) <-chan int {
out := make(chan int)
go func() {
defer close(out)
for n := range in {
if n > threshold {
out <- n
}
}
}()
return out
}
func main() {
// 构建 pipeline: generator -> square -> filter
nums := generator(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
squares := square(nums)
filtered := filter(squares, 20)
for n := range filtered {
fmt.Println(n) // 25, 36, 49, 64, 81, 100
}
}Go 的并发哲学是"不要通过共享内存来通信，而要通过通信来共享内存"。正确使用 channel 和 goroutine，能构建出既高效又安全的并发程序。
读者评论 2