Go09-并发编程

第8章：并发编程

8.1 并发基础

Go 使用 goroutine 和 channel 实现并发，这是 Go 最强大的特性之一。

Goroutine

Goroutine 是轻量级线程，由 Go 运行时管理：

packagemainimport("fmt""time")funcsayHello(namestring){fori:=0;i<5;i++{fmt.Printf("Hello, %s! (%d)\n",name,i)time.Sleep(100*time.Millisecond)}}funcmain(){// 启动 goroutinegosayHello("张三")gosayHello("李四")// 主 goroutine 等待time.Sleep(1*time.Second)fmt.Println("主程序结束")}

启动多个 Goroutine

funcmain(){// 启动多个 goroutinefori:=0;i<10;i++{gofunc(nint){fmt.Printf("goroutine %d\n",n)}(i)// 注意：传递参数避免闭包问题}time.Sleep(time.Second)}

8.2 Channel

Channel 是 goroutine 之间通信的管道。

基本 Channel

// 创建 channelch:=make(chanint)// 发送数据gofunc(){ch<-42// 发送}()// 接收数据value:=<-ch// 接收fmt.Println(value)// 42// 关闭 channelclose(ch)

带缓冲的 Channel

// 无缓冲 channel（同步）ch1:=make(chanint)// 带缓冲 channel（异步）ch2:=make(chanint,10)// 缓冲区大小为 10// 发送数据（不阻塞，直到缓冲区满）ch2<-1ch2<-2// 接收数据fmt.Println(<-ch2)// 1fmt.Println(<-ch2)// 2

Channel 方向

// 只发送 channelfuncproducer(chchan<-int){fori:=0;i<10;i++{ch<-i}close(ch)}// 只接收 channelfuncconsumer(ch<-chanint){forv:=rangech{fmt.Println(v)}}funcmain(){ch:=make(chanint)goproducer(ch)consumer(ch)}

8.3 Channel 操作

Select 语句

// 同时监听多个 channelfuncmain(){ch1:=make(chanstring)ch2:=make(chanstring)gofunc(){time.Sleep(1*time.Second)ch1<-"来自 channel 1"}()gofunc(){time.Sleep(2*time.Second)ch2<-"来自 channel 2"}()// select 等待第一个完成的 channelselect{casemsg:=<-ch1:fmt.Println(msg)casemsg:=<-ch2:fmt.Println(msg)case<-time.After(3*time.Second):fmt.Println("超时")}}

超时控制

funcfetchData()(string,error){ch:=make(chanstring,1)gofunc(){// 模拟耗时操作time.Sleep(2*time.Second)ch<-"数据"}()select{caseresult:=<-ch:returnresult,nilcase<-time.After(1*time.Second):return"",errors.New("请求超时")}}

非阻塞操作

select{casemsg:=<-ch:fmt.Println("收到消息:",msg)default:fmt.Println("没有消息")}

8.4 同步原语

sync.WaitGroup

packagemainimport("fmt""sync")funcmain(){varwg sync.WaitGroupfori:=0;i<10;i++{wg.Add(1)// 增加计数器gofunc(nint){deferwg.Done()// 减少计数器fmt.Printf("goroutine %d\n",n)}(i)}wg.Wait()// 等待所有 goroutine 完成fmt.Println("所有 goroutine 完成")}

sync.Mutex

packagemainimport("fmt""sync")typeCounterstruct{mu sync.Mutex valueint}func(c*Counter)Increment(){c.mu.Lock()// 加锁deferc.mu.Unlock()// 解锁c.value++}func(c*Counter)Value()int{c.mu.Lock()deferc.mu.Unlock()returnc.value}funcmain(){counter:=&Counter{}varwg sync.WaitGroupfori:=0;i<1000;i++{wg.Add(1)gofunc(){deferwg.Done()counter.Increment()}()}wg.Wait()fmt.Println("最终值:",counter.Value())// 1000}

sync.RWMutex（读写锁）

typeCachestruct{mu sync.RWMutex itemsmap[string]string}func(c*Cache)Get(keystring)(string,bool){c.mu.RLock()// 读锁deferc.mu.RUnlock()value,ok:=c.items[key]returnvalue,ok}func(c*Cache)Set(key,valuestring){c.mu.Lock()// 写锁deferc.mu.Unlock()c.items[key]=value}

sync.Once

typeSingletonstruct{instanceinterface{}once sync.Once}func(s*Singleton)Get()interface{}{s.once.Do(func(){s.instance=createInstance()})returns.instance}

sync.Map

packagemainimport("fmt""sync")funcmain(){varm sync.Map// 存储m.Store("key1","value1")m.Store("key2","value2")// 读取ifvalue,ok:=m.Load("key1");ok{fmt.Println(value)}// 遍历m.Range(func(key,valueinterface{})bool{fmt.Printf("%v: %v\n",key,value)returntrue// 返回 false 停止遍历})}

8.5 并发模式

Worker Pool

funcworker(idint,jobs<-chanint,resultschan<-int){forjob:=rangejobs{fmt.Printf("worker %d 处理 job %d\n",id,job)time.Sleep(time.Second)results<-job*2}}funcmain(){jobs:=make(chanint,100)results:=make(chanint,100)// 启动 workersforw:=1;w<=3;w++{goworker(w,jobs,results)}// 发送 jobsforj:=1;j<=9;j++{jobs<-j}close(jobs)// 收集结果fora:=1;a<=9;a++{fmt.Println("结果:",<-results)}}

Fan-out / Fan-in

// Fan-out: 多个 worker 读取同一个 channel// Fan-in: 多个 worker 写入同一个 channelfuncfanOut(input<-chanint,workersint)[]<-chanint{channels:=make([]<-chanint,workers)fori:=0;i<workers;i++{channels[i]=process(input)}returnchannels}funcfanIn(channels...<-chanint)<-chanint{varwg sync.WaitGroup merged:=make(chanint)for_,ch:=rangechannels{wg.Add(1)gofunc(c<-chanint){deferwg.Done()forv:=rangec{merged<-v}}(ch)}gofunc(){wg.Wait()close(merged)}()returnmerged}

Pipeline

funcgenerate(nums...int)<-chanint{out:=make(chanint)gofunc(){for_,n:=rangenums{out<-n}close(out)}()returnout}funcsquare(in<-chanint)<-chanint{out:=make(chanint)gofunc(){forn:=rangein{out<-n*n}close(out)}()returnout}funcmain(){// Pipeline: generate -> squareforv:=rangesquare(generate(1,2,3,4,5)){fmt.Println(v)// 1, 4, 9, 16, 25}}

8.6 练习

1. 实现一个简单的并发爬虫
2. 使用 Worker Pool 模式处理批量任务
3. 实现一个并发安全的缓存
4. 编写一个 Pipeline 处理数据流

8.7 下一章

下一章我们将学习 Go 语言的标准库精讲。