并发模式
并发模式简介
并发模式是经过验证的策略，用于管理和协调协程，以高效且安全地解决复杂的计算问题。
常见并发模式
1. 工作池模式
func workerPool(jobs <-chan int, results chan<- int, workerCount int) {
for i := 0; i < workerCount; i++ {
go func() {
for job := range jobs {
results <- processJob(job)
}
}()
}
}
func processJob(job int) int {
// 模拟复杂处理
time.Sleep(time.Millisecond)
return job * 2
}
func main() {
jobs := make(chan int, 100)
results := make(chan int, 100)
workerPool(jobs, results, 10)
// 发送任务
for i := 0; i < 50; i++ {
jobs <- i
}
close(jobs)
// 收集结果
for i := 0; i < 50; i++ {
<-results
}
}
2. 扇出/扇入模式
graph TD
A[输入通道] --> B[分发器]
B --> C1[工作者1]
B --> C2[工作者2]
B --> C3[工作者3]
C1 --> D[收集器]
C2 --> D
C3 --> D
func fanOutFanIn(input <-chan int) <-chan int {
numWorkers := runtime.NumCPU()
outputs := make([]<-chan int, numWorkers)
// 扇出
for i := 0; i < numWorkers; i++ {
outputs[i] = worker(input)
}
// 扇入
return merge(outputs...)
}
func worker(input <-chan int) <-chan int {
output := make(chan int)
go func() {
for num := range input {
output <- processNumber(num)
}
close(output)
}()
return output
}
func merge(channels...<-chan int) <-chan int {
var wg sync.WaitGroup
mergedChan := make(chan int)
output := func(c <-chan int) {
defer wg.Done()
for num := range c {
mergedChan <- num
}
}
wg.Add(len(channels))
for _, ch := range channels {
go output(ch)
}
go func() {
wg.Wait()
close(mergedChan)
}()
return mergedChan
}
同步模式
3. 信号量模式
type Semaphore struct {
sem chan struct{}
}
func NewSemaphore(maxConcurrency int) *Semaphore {
return &Semaphore{
sem: make(chan struct{}, maxConcurrency),
}
}
func (s *Semaphore) Acquire() {
s.sem <- struct{}{}
}
func (s *Semaphore) Release() {
<-s.sem
}
func main() {
sem := NewSemaphore(3)
var wg sync.WaitGroup
for i := 0; i < 10; i++ {
wg.Add(1)
go func(id int) {
defer wg.Done()
sem.Acquire()
defer sem.Release()
fmt.Printf("Processing task %d\n", id)
time.Sleep(time.Second)
}(i)
}
wg.Wait()
}
并发模式比较
| 模式 |
使用场景 |
优点 |
缺点 |
| 工作池 |
并行处理 |
可控并发 |
固定工作者数量 |
| 扇出/扇入 |
分布式计算 |
可扩展 |
实现复杂 |
| 信号量 |
资源限制 |
可控访问 |
可能死锁 |
高级模式
4. 管道模式
func pipeline() <-chan int {
out := make(chan int)
go func() {
for i := 1; i <= 10; i++ {
out <- i
}
close(out)
}()
return out
}
func square(in <-chan int) <-chan int {
out := make(chan int)
go func() {
for num := range in {
out <- num * num
}
close(out)
}()
return out
}
最佳实践
- 使用通道进行通信
- 尽量减少共享状态
- 设计支持取消
- 优雅处理错误
- 全面测试并发代码
实际考量
- 为特定用例选择合适的模式
- 考虑性能影响
- 使用 LabEx 的并发编程环境进行实践
- 分析和基准测试并发代码
结论
掌握并发模式对于编写高效、可扩展的 Go 应用程序至关重要。每种模式都解决特定的同步和协调挑战。
