Go进阶概览 -【5.5 并发编程中的常见模式与最佳实践】

5.5 并发编程中的常见模式与最佳实践

在现代编程中，并发编程是提高程序性能和响应速度的关键技术之一。

Go语言以其轻量级的协程和强大的并发支持，成为了处理并发任务的理想选择。

本节将介绍Go语言中常见的并发编程模式，并探讨如何编写高效可靠的并发代码。

本节代码存放目录为 lesson17

生产者-消费者模式

生产者-消费者模式是一种经典的并发模式，通常用于解决生产者与消费者速度不一致的问题。

生产者负责生成数据，而消费者负责处理数据。通过引入缓冲区（通常使用 channel），生产者和消费者可以并发工作，而无需相互等待。

实现代码如下所示：

func producer(ch chan<- int, wg *sync.WaitGroup) {defer wg.Done()for i := 0; i < 10; i++ {ch <- ifmt.Println("Produced: ", i)time.Sleep(time.Duration(1) * time.Second)}close(ch)
}func consumer(ch <-chan int, wg *sync.WaitGroup) {defer wg.Done()for item := range ch {fmt.Println("Consumed: ", item)}
}func main() {var (wg sync.WaitGroup)ch := make(chan int, 5)wg.Add(2)go producer(ch, &wg)go consumer(ch, &wg)wg.Wait()
}

使用注意点

使用close关闭channel，防止消费者在无数据可读时死锁。
使用缓冲通道提高生产者和消费者之间的处理效率。

工作池模式

工作池模式用于在处理大量任务时，限制同时执行的任务数量。

通过固定数量的Goroutine（工人），我们可以高效地处理大量任务，同时控制资源的使用。

实现代码如下所示：

func worker(id int, jobs <-chan int, results chan<- int, wg *sync.WaitGroup) {defer wg.Done()for j := range jobs {fmt.Printf("Worker %d started job %d\n", id, j)time.Sleep(time.Second)fmt.Printf("Worker %d finished job %d\n", id, j)results <- j * 2}
}func main() {var (wg sync.WaitGroup)jobs := make(chan int, 100)results := make(chan int, 100)for w := 1; w <= 3; w++ {wg.Add(1)go worker(w, jobs, results, &wg)}for j := 1; j <= 9; j++ {jobs <- j}close(jobs)wg.Wait()close(results)for r := range results {fmt.Println("Result: ", r)}
}

使用注意点

设置适当的Worker数量以平衡资源使用和任务处理速度。
使用sync.WaitGroup确保所有任务完成后程序再退出。

扇入扇出模式

扇入扇出模式常用于将多个输入源的结果汇总到一起，或将单个任务分解为多个并发子任务来处理，然后将结果汇总。

实现代码如下所示：

func process(id int, ch chan<- int) {defer close(ch)result := id * 2fmt.Printf("Process %d done\n", id)ch <- result
}func fanIn(channels ...<-chan int) <-chan int {out := make(chan int)var (wg sync.WaitGroup)wg.Add(len(channels))for _, ch := range channels {go func(c <-chan int) {defer wg.Done()for val := range c {out <- val}}(ch)}go func() {wg.Wait()close(out) // 确保所有 Goroutine 处理完毕后关闭输出通道}()return out
}func main() {channels := make([]chan int, 3)for i := range channels {channels[i] = make(chan int)go process(i+1, channels[i])}resultCh := fanIn(channels[0], channels[1], channels[2])for result := range resultCh {fmt.Println("Received:", result)}
}

使用注意点

确保所有输入channel都被关闭，防止数据泄漏或死锁。
使用扇入机制简化结果的汇总和处理。

超时与取消

在并发编程中，处理任务超时和取消非常重要，尤其是当某些任务可能无法在合理的时间内完成时。

实现代码如下所示：

func worker(ctx context.Context, id int, ch chan<- int) {select {case <-time.After(4 * time.Second):ch <- id * 2case <-ctx.Done():fmt.Printf("Worker %d canceled\n", id)}
}func main() {ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)defer cancel()ch := make(chan int)for i := 1; i <= 5; i++ {go worker(ctx, i, ch)}for i := 1; i <= 5; i++ {select {case res := <-ch:fmt.Println("Received: ", res)case <-ctx.Done():fmt.Println("Main canceled")return}}
}