Go语言中多阶段算法的并行化：利用Goroutine与缓冲通道构建高效数据管道

package main

import (
    "fmt"
    "sync"
    "time"
)

// 假设的中间数据类型
type RawStreamData struct{ id int }
type SymbolSequence struct{ id int }
type ImageFrame struct{ id int }
type OutputFormatData struct{ id int }

// 定义缓冲通道的容量
const bufferSize = 10

// Stage 1: 反序列化输入流
func deserializeStage(inputID int, rawDataChan chan<- RawStreamData, wg *sync.WaitGroup) {
    defer wg.Done()
    defer close(rawDataChan) // 完成后关闭通道
    fmt.Printf("Stage 1: 开始反序列化输入流...\n")
    for i := 0; i < inputID; i++ {
        data := RawStreamData{id: i}
        time.Sleep(time.Millisecond * 50) // 模拟处理时间
        rawDataChan <- data
        fmt.Printf("Stage 1: 反序列化数据 %d\n", data.id)
    }
    fmt.Printf("Stage 1: 反序列化完成。\n")
}

// Stage 2: 生成符号序列
func generateSymbolsStage(rawDataChan <-chan RawStreamData, symbolChan chan<- SymbolSequence, wg *sync.WaitGroup) {
    defer wg.Done()
    defer close(symbolChan) // 完成后关闭通道
    fmt.Printf("Stage 2: 开始生成符号序列...\n")
    for rawData := range rawDataChan {
        symbol := SymbolSequence{id: rawData.id}
        time.Sleep(time.Millisecond * 80) // 模拟处理时间
        symbolChan <- symbol
        fmt.Printf("Stage 2: 生成符号序列 %d\n", symbol.id)
    }
    fmt.Printf("Stage 2: 符号序列生成完成。\n")
}

// Stage 3: 生成图像流
func generateImagesStage(symbolChan <-chan SymbolSequence, imageChan chan<- ImageFrame, wg *sync.WaitGroup) {
    defer wg.Done()
    defer close(imageChan) // 完成后关闭通道
    fmt.Printf("Stage 3: 开始生成图像流...\n")
    for symbol := range symbolChan {
        image := ImageFrame{id: symbol.id}
        time.Sleep(time.Millisecond * 150) // 模拟处理时间，这是瓶颈之一
        imageChan <- image
        fmt.Printf("Stage 3: 生成图像 %d\n", image.id)
    }
    fmt.Printf("Stage 3: 图像流生成完成。\n")
}

// Stage 4: 序列化图像流
func serializeOutputStage(imageChan <-chan ImageFrame, wg *sync.WaitGroup) {
    defer wg.Done()
    fmt.Printf("Stage 4: 开始序列化输出...\n")
    for image := range imageChan {
        output := OutputFormatData{id: image.id}
        time.Sleep(time.Millisecond * 200) // 模拟处理时间，这是另一个瓶颈
        _ = output                          // 假设数据已被处理
        fmt.Printf("Stage 4: 序列化输出 %d\n", output.id)
    }
    fmt.Printf("Stage 4: 序列化输出完成。\n")
}

func main() {
    var wg sync.WaitGroup

    // 创建缓冲通道连接各个阶段
    rawDataChan := make(chan RawStreamData, bufferSize)
    symbolChan := make(chan SymbolSequence, bufferSize)
    imageChan := make(chan ImageFrame, bufferSize)

    // 启动各个阶段的Goroutine
    wg.Add(4)
    go deserializeStage(5, rawDataChan, &wg) // 假设处理5个数据单元
    go generateSymbolsStage(rawDataChan, symbolChan, &wg)
    go generateImagesStage(symbolChan, imageChan, &wg)
    go serializeOutputStage(imageChan, &wg)

    // 等待所有Goroutine完成
    wg.Wait()
    fmt.Println("所有处理阶段均已完成。")
}