第七章：并发编程
7.1 Goroutine
Goroutine基础
package main
import (
"fmt"
"time"
)
// 普通函数
func printNumbers() {
for i := 1; i <= 5; i++ {
fmt.Printf("数字: %d\\n", i)
time.Sleep(100 * time.Millisecond)
}
}
func printLetters() {
for ch := 'A'; ch <= 'E'; ch++ {
fmt.Printf("字母: %c\\n", ch)
time.Sleep(150 * time.Millisecond)
}
}
func main() {
// 顺序执行
fmt.Println("顺序执行:")
printNumbers()
printLetters()
// 并发执行
fmt.Println("\\n并发执行:")
go printNumbers() // 启动goroutine
go printLetters() // 启动goroutine
// 等待goroutine完成
time.Sleep(2 * time.Second)
fmt.Println("主函数结束")
}
WaitGroup
package main
import (
"fmt"
"sync"
"time"
)
func worker(id int, wg *sync.WaitGroup) {
defer wg.Done() // 完成时通知WaitGroup
fmt.Printf("Worker %d 开始工作\\n", id)
time.Sleep(time.Duration(id) * 100 * time.Millisecond)
fmt.Printf("Worker %d 完成工作\\n", id)
}
func main() {
var wg sync.WaitGroup
// 启动5个worker
for i := 1; i <= 5; i++ {
wg.Add(1) // 增加计数器
go worker(i, &wg)
}
// 等待所有worker完成
wg.Wait()
fmt.Println("所有worker已完成")
}
Mutex互斥锁
package main
import (
"fmt"
"sync"
)
// SafeCounter 线程安全的计数器
type SafeCounter struct {
mu sync.Mutex
count int
}
func (c *SafeCounter) Increment() {
c.mu.Lock()
defer c.mu.Unlock()
c.count++
}
func (c *SafeCounter) Decrement() {
c.mu.Lock()
defer c.mu.Unlock()
c.count—
}
func (c *SafeCounter) Value() int {
c.mu.Lock()
defer c.mu.Unlock()
return c.count
}
func main() {
counter := &SafeCounter{}
var wg sync.WaitGroup
// 启动100个goroutine递增
for i := 0; i < 100; i++ {
wg.Add(1)
go func() {
defer wg.Done()
counter.Increment()
}()
}
// 启动50个goroutine递减
for i := 0; i < 50; i++ {
wg.Add(1)
go func() {
defer wg.Done()
counter.Decrement()
}()
}
wg.Wait()
fmt.Printf("最终计数: %d\\n", counter.Value()) // 应该是50
}
RWMutex读写锁
package main
import (
"fmt"
"sync"
"time"
)
// SafeMap 线程安全的Map
type SafeMap struct {
mu sync.RWMutex
data map[string]string
}
func NewSafeMap() *SafeMap {
return &SafeMap{
data: make(map[string]string),
}
}
func (m *SafeMap) Get(key string) (string, bool) {
m.mu.RLock() // 读锁
defer m.mu.RUnlock()
val, ok := m.data[key]
return val, ok
}
func (m *SafeMap) Set(key, value string) {
m.mu.Lock() // 写锁
defer m.mu.Unlock()
m.data[key] = value
}
func (m *SafeMap) Delete(key string) {
m.mu.Lock()
defer m.mu.Unlock()
delete(m.data, key)
}
func (m *SafeMap) Len() int {
m.mu.RLock()
defer m.mu.RUnlock()
return len(m.data)
}
func main() {
sm := NewSafeMap()
var wg sync.WaitGroup
// 并发写入
for i := 0; i < 10; i++ {
wg.Add(1)
go func(id int) {
defer wg.Done()
key := fmt.Sprintf("key%d", id)
value := fmt.Sprintf("value%d", id)
sm.Set(key, value)
fmt.Printf("写入: %s = %s\\n", key, value)
}(i)
}
// 并发读取
for i := 0; i < 10; i++ {
wg.Add(1)
go func(id int) {
defer wg.Done()
key := fmt.Sprintf("key%d", id)
time.Sleep(50 * time.Millisecond)
if val, ok := sm.Get(key); ok {
fmt.Printf("读取: %s = %s\\n", key, val)
}
}(i)
}
wg.Wait()
fmt.Printf("Map大小: %d\\n", sm.Len())
}
7.2 Channel
Channel基础
package main
import "fmt"
func main() {
// 创建channel
ch := make(chan int)
// 发送数据到channel
go func() {
ch <- 42 // 发送
}()
// 从channel接收数据
value := <-ch // 接收
fmt.Printf("接收到: %d\\n", value)
// 带缓冲的channel
bufferedCh := make(chan string, 3)
bufferedCh <- "Hello"
bufferedCh <- "World"
bufferedCh <- "Go"
fmt.Println(<-bufferedCh) // Hello
fmt.Println(<-bufferedCh) // World
fmt.Println(<-bufferedCh) // Go
}
Channel方向
package main
import "fmt"
// 只写channel
func producer(ch chan<- int) {
for i := 0; i < 5; i++ {
ch <- i
}
close(ch)
}
// 只读channel
func consumer(ch <-chan int) {
for value := range ch {
fmt.Printf("消费: %d\\n", value)
}
}
func main() {
ch := make(chan int)
go producer(ch)
consumer(ch)
}
Select语句
package main
import (
"fmt"
"time"
)
func main() {
ch1 := make(chan string)
ch2 := make(chan string)
go func() {
time.Sleep(1 * time.Second)
ch1 <- "来自ch1"
}()
go func() {
time.Sleep(2 * time.Second)
ch2 <- "来自ch2"
}()
// 使用select等待多个channel
for i := 0; i < 2; i++ {
select {
case msg1 := <-ch1:
fmt.Println(msg1)
case msg2 := <-ch2:
fmt.Println(msg2)
}
}
// select超时控制
ch := make(chan string)
go func() {
time.Sleep(3 * time.Second)
ch <- "数据"
}()
select {
case msg := <-ch:
fmt.Println("收到:", msg)
case <-time.After(2 * time.Second):
fmt.Println("超时!")
}
// 非阻塞select
ch3 := make(chan int, 1)
select {
case v := <-ch3:
fmt.Println("收到:", v)
default:
fmt.Println("没有数据")
}
ch3 <- 42
select {
case v := <-ch3:
fmt.Println("收到:", v)
default:
fmt.Println("没有数据")
}
}
Channel模式
package main
import (
"fmt"
"sync"
)
// 扇出模式 (Fan-out)
func fanOut(input <-chan int, workers int) []<-chan int {
channels := make([]<-chan int, workers)
for i := 0; i < workers; i++ {
channels[i] = worker(input, i)
}
return channels
}
func worker(input <-chan int, id int) <-chan int {
output := make(chan int)
go func() {
defer close(output)
for v := range input {
result := v * v // 处理数据
fmt.Printf("Worker %d: %d -> %d\\n", id, v, result)
output <- result
}
}()
return output
}
// 扇入模式 (Fan-in)
func fanIn(channels …<-chan int) <-chan int {
var wg sync.WaitGroup
merged := make(chan int)
// 为每个channel启动一个goroutine
wg.Add(len(channels))
for _, ch := range channels {
go func(c <-chan int) {
defer wg.Done()
for v := range c {
merged <- v
}
}(ch)
}
// 等待所有channel关闭后关闭merged
go func() {
wg.Wait()
close(merged)
}()
return merged
}
// 管道模式 (Pipeline)
func pipeline() {
// 第一阶段: 生成数据
nums := make(chan int)
go func() {
defer close(nums)
for i := 1; i <= 10; i++ {
nums <- i
}
}()
// 第二阶段: 平方
squares := make(chan int)
go func() {
defer close(squares)
for n := range nums {
squares <- n * n
}
}()
// 第三阶段: 过滤偶数
evens := make(chan int)
go func() {
defer close(evens)
for n := range squares {
if n%2 == 0 {
evens <- n
}
}
}()
// 消费结果
for n := range evens {
fmt.Println(n)
}
}
func main() {
fmt.Println("管道模式示例:")
pipeline()
fmt.Println("\\n扇出-扇入模式示例:")
// 创建输入channel
input := make(chan int)
go func() {
defer close(input)
for i := 1; i <= 10; i++ {
input <- i
}
}()
// 扇出: 多个worker处理
channels := fanOut(input, 3)
// 扇入: 合并结果
merged := fanIn(channels…)
// 收集结果
results := make([]int, 0)
for v := range merged {
results = append(results, v)
}
fmt.Printf("结果: %v\\n", results)
}
7.3 Context
Context基础
package main
import (
"context"
"fmt"
"time"
)
func worker(ctx context.Context, id int) {
for {
select {
case <-ctx.Done():
fmt.Printf("Worker %d: 收到取消信号, 原因: %v\\n", id, ctx.Err())
return
default:
fmt.Printf("Worker %d: 正在工作…\\n", id)
time.Sleep(500 * time.Millisecond)
}
}
}
func main() {
// 创建可取消的context
ctx, cancel := context.WithCancel(context.Background())
// 启动多个worker
for i := 1; i <= 3; i++ {
go worker(ctx, i)
}
// 3秒后取消所有worker
time.Sleep(3 * time.Second)
fmt.Println("主函数: 发送取消信号")
cancel()
// 等待worker退出
time.Sleep(1 * time.Second)
fmt.Println("主函数: 结束")
}
Context超时和值
package main
import (
"context"
"fmt"
"time"
)
// 模拟HTTP请求处理
func handleRequest(ctx context.Context) {
// 从context获取值
userID, ok := ctx.Value("userID").(string)
if ok {
fmt.Printf("处理用户 %s 的请求\\n", userID)
}
// 模拟长时间操作
select {
case <-time.After(3 * time.Second):
fmt.Println("请求处理完成")
case <-ctx.Done():
fmt.Printf("请求被取消: %v\\n", ctx.Err())
}
}
func main() {
// 超时context
fmt.Println("示例1: 超时context")
ctx1, cancel1 := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel1()
go handleRequest(ctx1)
time.Sleep(3 * time.Second)
// 带截止时间的context
fmt.Println("\\n示例2: 截止时间context")
deadline := time.Now().Add(2 * time.Second)
ctx2, cancel2 := context.WithDeadline(context.Background(), deadline)
defer cancel2()
go handleRequest(ctx2)
time.Sleep(3 * time.Second)
// 带值的context
fmt.Println("\\n示例3: 带值的context")
ctx3 := context.WithValue(context.Background(), "userID", "user123")
ctx3, cancel3 := context.WithTimeout(ctx3, 2*time.Second)
defer cancel3()
go handleRequest(ctx3)
time.Sleep(3 * time.Second)
// context链
fmt.Println("\\n示例4: context链")
parentCtx, parentCancel := context.WithCancel(context.Background())
childCtx, childCancel := context.WithTimeout(parentCtx, 5*time.Second)
defer parentCancel()
defer childCancel()
go func() {
<-childCtx.Done()
fmt.Printf("子context: %v\\n", childCtx.Err())
}()
// 取消父context会影响子context
time.Sleep(1 * time.Second)
parentCancel()
time.Sleep(1 * time.Second)
}
7.4 实战案例：并发下载器
package main
import (
"context"
"fmt"
"math/rand"
"strings"
"sync"
"time"
)
// DownloadTask 下载任务
type DownloadTask struct {
ID int
URL string
}
// DownloadResult 下载结果
type DownloadResult struct {
TaskID int
URL string
Size int
Duration time.Duration
Error error
}
// Downloader 并发下载器
type Downloader struct {
maxWorkers int
tasks chan DownloadTask
results chan DownloadResult
ctx context.Context
cancel context.CancelFunc
wg sync.WaitGroup
}
// NewDownloader 创建下载器
func NewDownloader(maxWorkers int) *Downloader {
ctx, cancel := context.WithCancel(context.Background())
return &Downloader{
maxWorkers: maxWorkers,
tasks: make(chan DownloadTask, 100),
results: make(chan DownloadResult, 100),
ctx: ctx,
cancel: cancel,
}
}
// Start 启动下载器
func (d *Downloader) Start() {
// 启动worker池
for i := 0; i < d.maxWorkers; i++ {
d.wg.Add(1)
go d.worker(i)
}
// 收集结果
go d.collectResults()
}
// worker 工作协程
func (d *Downloader) worker(id int) {
defer d.wg.Done()
for {
select {
case <-d.ctx.Done():
fmt.Printf("Worker %d: 收到停止信号\\n", id)
return
case task, ok := <-d.tasks:
if !ok {
fmt.Printf("Worker %d: 任务通道已关闭\\n", id)
return
}
d.processTask(id, task)
}
}
}
// processTask 处理下载任务
func (d *Downloader) processTask(workerID int, task DownloadTask) {
start := time.Now()
// 模拟下载过程
duration := time.Duration(rand.Intn(3)+1) * time.Second
time.Sleep(duration)
// 模拟随机失败
var err error
if rand.Float32() < 0.1 { // 10%失败率
err = fmt.Errorf("下载失败: 网络错误")
}
size := rand.Intn(1000) + 100
result := DownloadResult{
TaskID: task.ID,
URL: task.URL,
Size: size,
Duration: time.Since(start),
Error: err,
}
d.results <- result
}
// collectResults 收集结果
func (d *Downloader) collectResults() {
for result := range d.results {
if result.Error != nil {
fmt.Printf("任务 %d 失败: %v\\n", result.TaskID, result.Error)
} else {
fmt.Printf("任务 %d 完成: %s (%dKB, 耗时%v)\\n",
result.TaskID, result.URL, result.Size, result.Duration)
}
}
}
// AddTask 添加下载任务
func (d *Downloader) AddTask(task DownloadTask) {
d.tasks <- task
}
// Stop 停止下载器
func (d *Downloader) Stop() {
close(d.tasks)
d.wg.Wait()
close(d.results)
d.cancel()
}
// DownloadManager 下载管理器
type DownloadManager struct {
downloader *Downloader
totalTasks int
completed int
failed int
mu sync.Mutex
}
// NewDownloadManager 创建管理器
func NewDownloadManager(maxWorkers int) *DownloadManager {
return &DownloadManager{
downloader: NewDownloader(maxWorkers),
}
}
// Start 启动管理器
func (dm *DownloadManager) Start() {
dm.downloader.Start()
}
// AddURL 添加URL
func (dm *DownloadManager) AddURL(url string) {
dm.mu.Lock()
dm.totalTasks++
dm.mu.Unlock()
dm.downloader.AddTask(DownloadTask{
ID: dm.totalTasks,
URL: url,
})
}
// Stop 停止管理器
func (dm *DownloadManager) Stop() {
dm.downloader.Stop()
}
func main() {
// 创建下载管理器
manager := NewDownloadManager(3) // 3个worker
// 启动
manager.Start()
// 添加下载任务
urls := []string{
"https://example.com/file1.zip",
"https://example.com/file2.zip",
"https://example.com/file3.zip",
"https://example.com/file4.zip",
"https://example.com/file5.zip",
"https://example.com/file6.zip",
"https://example.com/file7.zip",
"https://example.com/file8.zip",
"https://example.com/file9.zip",
"https://example.com/file10.zip",
}
for _, url := range urls {
manager.AddURL(url)
}
// 等待完成
time.Sleep(10 * time.Second)
// 停止
manager.Stop()
fmt.Println("\\n下载完成!")
}
7.5 sync包常用工具
sync.Once
package main
import (
"fmt"
"sync"
)
var once sync.Once
var config map[string]string
func loadConfig() {
fmt.Println("加载配置…")
config = map[string]string{
"host": "localhost",
"port": "8080",
}
}
func GetConfig() map[string]string {
once.Do(loadConfig) // 只执行一次
return config
}
func main() {
var wg sync.WaitGroup
// 多个goroutine同时获取配置
for i := 0; i < 5; i++ {
wg.Add(1)
go func(id int) {
defer wg.Done()
cfg := GetConfig()
fmt.Printf("Goroutine %d: %v\\n", id, cfg)
}(i)
}
wg.Wait()
}
sync.Pool
package main
import (
"fmt"
"sync"
)
type Buffer struct {
Data []byte
}
func NewBuffer() *Buffer {
fmt.Println("创建新Buffer")
return &Buffer{
Data: make([]byte, 0, 1024),
}
}
func (b *Buffer) Reset() {
b.Data = b.Data[:0]
}
func main() {
// 创建对象池
pool := sync.Pool{
New: func() interface{} {
return NewBuffer()
},
}
// 从池中获取对象
buf1 := pool.Get().(*Buffer)
buf1.Data = append(buf1.Data, "Hello"…)
fmt.Printf("buf1: %s\\n", buf1.Data)
// 归还到池中
buf1.Reset()
pool.Put(buf1)
// 再次获取（可能重用之前的对象）
buf2 := pool.Get().(*Buffer)
fmt.Printf("buf2: %s (长度: %d)\\n", buf2.Data, len(buf2.Data))
// 并发使用
var wg sync.WaitGroup
for i := 0; i < 10; i++ {
wg.Add(1)
go func(id int) {
defer wg.Done()
buf := pool.Get().(*Buffer)
buf.Data = append(buf.Data, fmt.Sprintf("goroutine-%d", id)…)
fmt.Printf("Goroutine %d: %s\\n", id, buf.Data)
buf.Reset()
pool.Put(buf)
}(i)
}
wg.Wait()
}
sync.Map
package main
import (
"fmt"
"sync"
)
func main() {
var m sync.Map
// 存储
m.Store("name", "张三")
m.Store("age", 25)
m.Store("city", "北京")
// 读取
name, ok := m.Load("name")
if ok {
fmt.Printf("name: %v\\n", name)
}
// 存储或获取
actual, loaded := m.LoadOrStore("name", "李四")
fmt.Printf("actual: %v, loaded: %v\\n", actual, loaded)
// 删除
m.Delete("age")
// 遍历
m.Range(func(key, value interface{}) bool {
fmt.Printf("%v: %v\\n", key, value)
return true
})
// 并发安全
var wg sync.WaitGroup
for i := 0; i < 10; i++ {
wg.Add(1)
go func(id int) {
defer wg.Done()
key := fmt.Sprintf("key%d", id)
m.Store(key, id)
}(i)
}
wg.Wait()
fmt.Println("\\n并发写入后的数据:")
m.Range(func(key, value interface{}) bool {
fmt.Printf("%v: %v\\n", key, value)
return true
})
}
