承上启下
我们在之前已经学习了goroutine和channel的并发模式，也学会了sync库和context的控制。那么在Go里面一般都会使用哪些高并发模式呢？今天让我们在这篇文章中一起揭晓一下。
开始学习
for ... select...模式
for select模式是Go语言中处理并发的一种常见模式，它结合了for循环和select语句来处理多个channel上的通信。这种模式通常用于以下场景：
- 同时从多个channel接收数据。
- 在一个循环中处理多个channel上的消息，直到满足某个条件（例如接收到一个特殊的消息或达到某个截止时间）。
下面是for select模式的基本结构和用法：
基本结构
for {
select {
case <-ch1:
// 处理从ch1接收到的数据
case data := <-ch2:
// 处理从ch2接收到的数据
case ch3 <- data:
// 向ch3发送数据
default:
// 当没有其他case准备就绪时执行
}
// 可能还有其他逻辑
// 可以在这里检查是否应该退出循环
}
用法说明
-
接收数据：在
select的case子句中，你可以从多个channel接收数据。当任何一个channel准备好发送数据时，相应的case就会被执行。 -
发送数据：你也可以在
select的case子句中向channel发送数据。如果channel准备好接收数据，则case会被执行。 -
默认情况：
default子句是可选的。如果没有channel准备好通信，并且提供了default子句，则default会被执行。这可以用来防止select无限期地阻塞，如果所有channel都没有准备好。 -
退出循环：通常在
for循环内部，你需要检查某些条件以决定是否应该退出循环。这可以通过设置一个标志、使用context包来处理取消信号，或者在接收到特定消息时完成。
select .. timeout
select timeout模式是Go语言中处理并发时的一种常见模式，它通过在select语句中包含一个超时情况来防止goroutine无限期地等待channel操作。这种模式对于避免goroutine因等待某些可能永远不会发生的事件而永久阻塞非常有用。
基本结构
下面是select timeout模式的基本结构：
select {
case <-ch:
// 当从ch接收到数据时执行
// 处理接收到的数据
case <-time.After(timeoutDuration):
// 当超时时间到达时执行
// 处理超时情况
}
用法说明
-
channel操作：
select的第一个case是正常的channel操作，可以是接收或发送数据。 -
超时操作：第二个
case使用了time.After函数，它返回一个在指定的持续时间timeoutDuration后发送当前时间的channel。当这个case被选中时，意味着已经超过了指定的超时时间。 -
处理超时：当超时发生时，你可以执行一些清理工作、设置错误状态、重试操作或者直接退出goroutine。
示例
以下是一个使用select timeout模式的示例，它展示了如何在等待一个channel操作时设置超时：
package main
import (
"fmt"
"time"
)
func main() {
ch := make(chan int)
timeoutDuration := 2 * time.Second
// 启动一个goroutine来模拟一个可能延迟的操作
go func() {
time.Sleep(3 * time.Second) // 模拟耗时操作
ch <- 42 // 发送数据
}()
// 使用select timeout模式等待channel操作或超时
select {
case data := <-ch:
fmt.Println("Received data:", data)
case <-time.After(timeoutDuration):
fmt.Println("Operation timed out")
}
fmt.Println("Done.")
}
在这个示例中，我们创建了一个channel ch 和一个超时时间 timeoutDuration。然后，我们启动一个goroutine来模拟一个耗时操作，该操作在3秒后向channel发送一个值。主goroutine使用select timeout模式来等待从channel接收数据或超时。由于我们设置的超时时间为2秒，而goroutine需要3秒来发送数据，因此会触发超时case。
运行这个程序，你会看到输出：
Operation timed out
Done.
这个模式确保了即使channel操作没有在预期的时间内完成，程序也不会无限期地等待，而是可以采取适当的超时处理措施。
Pipeline模式
Pipeline模式在Go语言中是一种处理数据流的高效并发模式，它通过将数据处理任务分解成一系列的阶段（stages），每个阶段由一组goroutine组成，这些goroutine之间通过channel连接。这种模式类似于工厂流水线，每个工人（goroutine）负责一部分工作，然后将半成品传递给下一个工人。
基本结构
Pipeline模式通常包含以下三个主要部分：
-
生成器（Generator）：负责生成数据，通常是一个或多个goroutine，它们将数据发送到channel。
-
处理器（Processor）：负责处理数据，可以是多个goroutine，每个goroutine从输入channel接收数据，处理完毕后发送到输出channel。
-
汇聚器（Collector）：负责收集最终结果，通常是一个或多个goroutine，它们从channel接收数据并执行最终的聚合操作。
以下是一个基本的Pipeline模式的示例：
package main
import "fmt"
// Generator: 生成0到9的数字
func generate(nums ...int) <-chan int {
out := make(chan int)
go func() {
for _, n := range nums {
out <- n
}
close(out)
}()
return out
}
// Processor: 将输入的数字乘以2
func multiply(in <-chan int) <-chan int {
out := make(chan int)
go func() {
for n := range in {
out <- n * 2
}
close(out)
}()
return out
}
// Collector: 打印结果
func print(in <-chan int) {
for n := range in {
fmt.Println(n)
}
}
func main() {
// 创建Pipeline
c := generate(0, 1, 2, 3, 4, 5, 6, 7, 8, 9)
out := multiply(c)
print(out)
}
在这个例子中，generate函数是一个生成器，它创建了一个goroutine来发送一系列数字。multiply函数是一个处理器，它创建了一个goroutine来接收这些数字，将它们乘以2，然后发送到另一个channel。最后，print函数作为汇聚器，它接收处理过的数字并打印它们。
特点
-
并行性：每个阶段可以并行处理数据，这提高了程序的并发性能。
-
解耦：每个阶段只关注自己的任务，并通过channel与其他阶段通信，降低了组件间的耦合。
-
灵活性：可以很容易地向Pipeline中添加或移除阶段，只需确保每个阶段的输入输出channel匹配。
-
可扩展性：可以通过增加更多的goroutine来扩展Pipeline的每个阶段，以处理更大的数据量。
Pipeline模式在Go语言中非常流行，因为它可以充分利用Go的并发特性来构建高效的数据处理流程。它适用于多种场景，如数据流处理、日志分析、图像处理等。
扇入和扇出模式
扇入（Fan-in）模式
扇入模式是指将多个子任务的结果汇总到一个channel中。在Go语言中，这通常是通过让多个goroutine将结果发送到同一个channel来实现的。
基本结构
func processTask(task Task, results chan<- Result) {
// 处理任务并将结果发送到channel
result := Result{}
// ... 计算结果 ...
results <- result
}
func fanIn(tasks []Task) []Result {
numTasks := len(tasks)
results := make(chan Result, numTasks) // 带缓冲的channel
var wg sync.WaitGroup
for _, task := range tasks {
wg.Add(1)
go func(t Task) {
defer wg.Done()
processTask(t, results)
}(task)
}
wg.Wait()
close(results)
var resultsSlice []Result
for result := range results {
resultsSlice = append(resultsSlice, result)
}
return resultsSlice
}
使用说明
- 结果收集：每个工作goroutine将处理结果发送到同一个channel。
- 缓冲channel：使用带缓冲的channel来减少发送操作的阻塞。
- 结果汇总：在所有工作goroutine完成后，关闭channel并从channel中读取所有结果。
在实际应用中，扇入和扇出模式通常结合使用。首先，使用扇出模式并行处理多个任务，然后使用扇入模式收集这些任务的结果。
示例
以下是一个结合使用扇入和扇出模式的示例：
package main
import (
"fmt"
"sync"
)
// 假设Task和Result是定义好的类型
type Task int
type Result int
func processTask(task Task, results chan<- Result) {
// 模拟任务处理
result := Result(task * 2)
results <- result
}
func fanOutFanIn(tasks []Task) []Result {
numTasks := len(tasks)
results := make(chan Result, numTasks)
var wg sync.WaitGroup
for _, task := range tasks {
wg.Add(1)
go func(t Task) {
defer wg.Done()
processTask(t, results)
}(task)
}
wg.Wait()
close(results)
var resultsSlice []Result
for result := range results {
resultsSlice = append(resultsSlice, result)
}
return resultsSlice
}
func main() {
tasks := []Task{1, 2, 3, 4, 5}
results := fanOutFanIn(tasks)
fmt.Println(results)
}
在这个例子中，fanOutFanIn函数首先使用扇出模式将任务分配给多个goroutine，然后使用扇入模式收集这些goroutine的结果。每个goroutine将处理结果发送到同一个channel，主goroutine等待所有goroutine完成后关闭channel，并从channel中读取所有结果。
Futures模式
Future模式是一种并发编程模式，它允许程序在等待某个操作完成的同时继续执行其他任务。在Future模式中，一个操作（通常是耗时的）被提交执行，并立即返回一个future对象，这个对象代表了操作的结果。其他部分的程序可以继续执行，而不必等待操作完成。当需要操作的结果时，程序可以检查future对象，如果操作已完成，则直接获取结果；如果操作尚未完成，则可以阻塞等待结果。
在Go语言中，Future模式可以通过以下方式实现：
使用Channel实现Future模式
在Go中，channel经常被用来实现Future模式。以下是一个简单的例子：
package main
import (
"fmt"
"time"
)
// AsyncFunction 启动一个goroutine来执行一个耗时的操作，并返回一个channel用于获取结果
func AsyncFunction() <-chan int {
resultChan := make(chan int)
go func() {
// 模拟耗时操作
time.Sleep(2 * time.Second)
resultChan <- 42 // 将结果发送到channel
}()
return resultChan
}
func main() {
// 调用AsyncFunction，它立即返回一个channel
future := AsyncFunction()
// 在等待结果的同时，可以执行其他操作
fmt.Println("Do some other work...")
// 当需要结果时，从channel读取
result := <-future
fmt.Printf("The result is: %d\n", result)
}
在上面的代码中，AsyncFunction函数启动了一个goroutine来执行一个耗时的操作，并返回一个channel。主goroutine可以在等待结果的同时执行其他任务。当需要结果时，它从channel中读取。
使用sync包实现Future模式
Go语言的sync包提供了一个sync.WaitGroup类型，可以用来等待一组操作完成。虽然它不直接提供future对象，但可以用来实现类似的功能。
package main
import (
"fmt"
"sync"
"time"
)
// DoWork 启动一个goroutine来执行一个耗时的操作，并通过WaitGroup等待其完成
func DoWork(wg *sync.WaitGroup, result *int) {
defer wg.Done()
// 模拟耗时操作
time.Sleep(2 * time.Second)
*result = 42 // 设置结果
}
func main() {
var wg sync.WaitGroup
var result int
wg.Add(1) // 增加WaitGroup的计数
go DoWork(&wg, &result) // 启动goroutine
// 在等待结果的同时，可以执行其他操作
fmt.Println("Do some other work...")
wg.Wait() // 等待goroutine完成
fmt.Printf("The result is: %d\n", result)
}
在这个例子中，DoWork函数在一个新的goroutine中执行耗时的操作，并通过指针参数返回结果。sync.WaitGroup用于等待goroutine完成。在DoWork执行的同时，主goroutine可以执行其他任务。当WaitGroup.Wait()被调用时，主goroutine会阻塞，直到DoWork完成。
这两种方法都可以在Go中实现Future模式，让你能够编写更高效的并发程序。使用Future模式，你可以有效地利用并发性，避免不必要的等待，从而提高程序的响应性和吞吐量。