我希望你能将这个流式写入脚本的复杂度并修复最后writeWg死锁的问题

package main

import (
	"context"
	"errors"
	"fmt"
	"sync"
	"time"

	"github.com/xuri/excelize/v2"
)

// Task 定义任务的数据结构
type Task struct {
	ID             int                    // 批次任务ID
	BatchRowIndex  int                    // 批次内的行索引
	GlobalRowIndex int                    // 全局行索引（0-9999）
	Data           map[string]interface{} // 输入数据
	Result         map[string]interface{} // 处理结果
	Retries        int                    // 重试次数
}

// TaskFunc 定义任务处理函数类型，返回单个结果
type TaskFunc func(ctx context.Context, data interface{}) (interface{}, error)

// Layer 定义流水线的每一层
type Layer struct {
	workerCount int
	timeout     time.Duration
	maxRetries  int
	taskFunc    TaskFunc
}

// Pipeline 定义整个流水线结构
type Pipeline struct {
	layers []*Layer
}

// NewPipeline 创建一个新的流水线实例
func NewPipeline() *Pipeline {
	return &Pipeline{
		layers: make([]*Layer, 0),
	}
}

// AddLayer 向流水线中添加一层
func (p *Pipeline) AddLayer(taskFunc TaskFunc, workerCount int, timeout time.Duration, maxRetries int) {
	layer := &Layer{
		workerCount: workerCount,
		timeout:     timeout,
		maxRetries:  maxRetries,
		taskFunc:    taskFunc,
	}
	p.layers = append(p.layers, layer)
}

// Run 启动流水线处理任务，并在每个任务完成时调用回调函数
func (p *Pipeline) Run(input []interface{}, callback func(task *Task)) error {
	if len(p.layers) == 0 {
		return errors.New("pipeline has no layers")
	}

	// 初始化任务通道，预估总任务数（10000行数据）
	inCh := make(chan *Task, len(input)*2000)

	// 将输入数据封装成任务并发送到第一个通道
	go func() {
		for _, data := range input {
			// 假设每个输入是一个批量数据（一个切片）
			batchData, ok := data.([]map[string]interface{})
			if !ok {
				fmt.Printf("Invalid batch data format: %v\n", data)
				continue
			}
			for i, row := range batchData {
				globalRowIndex := i
				task := &Task{
					ID:             0, // 如果有多个批次，调整此值
					BatchRowIndex:  globalRowIndex % 2000,
					GlobalRowIndex: globalRowIndex,
					Data:           row,
					Result:         nil,
					Retries:        0,
				}
				inCh <- task
			}
		}
		close(inCh)
	}()

	// 设置初始输入
	currentIn := inCh
	var currentOut chan *Task

	// 用于取消整个流水线的上下文（仅在严重错误时使用）
	ctxPipeline, cancelPipeline := context.WithCancel(context.Background())
	defer cancelPipeline()

	// 构建流水线
	for _, layer := range p.layers {
		currentOut = make(chan *Task, len(input)*2000)
		var wg sync.WaitGroup

		// 启动工作协程
		for w := 0; w < layer.workerCount; w++ {
			wg.Add(1)
			go func(layer *Layer, in <-chan *Task, out chan<- *Task) {
				defer wg.Done()
				for task := range in {
					retries := task.Retries
					var result interface{}
					var err error

					for {
						// 创建带超时的上下文
						ctx, cancel := context.WithTimeout(ctxPipeline, layer.timeout)
						result, err = layer.taskFunc(ctx, task.Data)
						cancel()

						if err != nil {
							retries++
							if retries > layer.maxRetries {
								// 超过重试次数，记录失败
								fmt.Printf("Task GlobalRowIndex %d failed after %d retries: %v\n", task.GlobalRowIndex, retries-1, err)
								task.Result = nil
								break
							}
							task.Retries = retries
							fmt.Printf("Task GlobalRowIndex %d retry %d due to error: %v\n", task.GlobalRowIndex, retries, err)
							continue
						}

						// 断言结果类型
						rowData, ok := result.(map[string]interface{})
						if !ok {
							fmt.Printf("Task GlobalRowIndex %d has invalid result type.\n", task.GlobalRowIndex)
							task.Result = nil
							err = errors.New("invalid result type")
							break
						}
						task.Result = rowData
						break
					}

					// 传递任务到下一层或回调
					out <- task
				}
			}(layer, currentIn, currentOut)
		}

		// 等待所有工作协程完成后关闭输出通道
		go func(wg *sync.WaitGroup, out chan *Task) {
			wg.Wait()
			close(out)
		}(&wg, currentOut)

		// 准备下一层的输入
		currentIn = currentOut
	}

	// 定义回调函数包装器，确保有序写入
	writer := callbackWrapper(callback)

	// 最后一层输出处理
	var wgFinal sync.WaitGroup
	wgFinal.Add(1)

	go func() {
		defer wgFinal.Done()
		for task := range currentOut {
			if task.Result == nil {
				// 处理失败任务
				fmt.Printf("Task GlobalRowIndex %d failed during processing.\n", task.GlobalRowIndex)
				// 根据需要处理失败任务，如记录日志或继续
				continue
			}
			// 调用回调函数处理每个任务的结果
			writer(task)
		}
	}()

	// 等待所有任务完成
	wgFinal.Wait()

	return nil
}

// callbackWrapper 确保按 GlobalRowIndex 顺序调用回调函数
func callbackWrapper(callback func(task *Task)) func(task *Task) {
	var (
		mu              sync.Mutex
		nextExpectedRow int
		buffer          = make(map[int]*Task)
	)

	return func(task *Task) {
		mu.Lock()
		defer mu.Unlock()

		if task.GlobalRowIndex == nextExpectedRow {
			// 写入当前任务
			callback(task)
			nextExpectedRow++

			// 检查缓冲区中是否有后续任务
			for {
				nextTask, exists := buffer[nextExpectedRow]
				if exists {
					callback(nextTask)
					delete(buffer, nextExpectedRow)
					nextExpectedRow++
				} else {
					break
				}
			}
		} else {
			// 缓存当前任务，等待前面的任务完成
			buffer[task.GlobalRowIndex] = task
		}
	}
}

// ExcelWriter 定义Excel写入器，使用 StreamWriter 进行流式写入
type ExcelWriter struct {
	file      *excelize.File
	sheetName string
	stream    *excelize.StreamWriter
	mutex     sync.Mutex // 保护写操作
}

// NewExcelWriter 初始化一个新的 ExcelWriter
func NewExcelWriter(sheetName string) (*ExcelWriter, error) {
	f := excelize.NewFile()

	// 创建新的工作表
	index, err := f.NewSheet(sheetName)
	if err != nil {
		return nil, err
	}
	f.SetActiveSheet(index)

	// 初始化表头
	headers := []interface{}{"A", "B", "C", "D", "E"} // 根据需要调整列
	sw, err := f.NewStreamWriter(sheetName)
	if err != nil {
		return nil, err
	}

	// 写入表头
	if err := sw.SetRow("A1", headers); err != nil {
		return nil, err
	}

	return &ExcelWriter{
		file:      f,
		sheetName: sheetName,
		stream:    sw,
	}, nil
}

// WriteRow 有序地写入一行数据到 Excel
func (ew *ExcelWriter) WriteRow(taskID int, globalRowIndex int, data map[string]interface{}) error {
	ew.mutex.Lock()
	defer ew.mutex.Unlock()

	// 计算 Excel 行号（假设行索引从 0 开始，表头在第 1 行）
	rowNum := globalRowIndex + 2
	cellName := fmt.Sprintf("A%d", rowNum)

	// 定义列顺序
	columns := []string{"A", "B", "C", "D", "E"}
	row := make([]interface{}, len(columns))
	for i, col := range columns {
		if cell, exists := data[col]; exists {
			row[i] = cell
		} else {
			row[i] = ""
		}
	}

	// 写入行
	if err := ew.stream.SetRow(cellName, row); err != nil {
		return err
	}

	return nil
}

// Save 保存并关闭 Excel 文件
func (ew *ExcelWriter) Save(filename string) error {
	ew.mutex.Lock()
	defer ew.mutex.Unlock()

	// 刷新所有缓冲数据
	if err := ew.stream.Flush(); err != nil {
		return err
	}

	// 保存文件
	return ew.file.SaveAs(filename)
}

// SplitTask 拆分批量任务为单行任务
func SplitTask(ctx context.Context, data interface{}) (interface{}, error) {
	// 这里数据已经是单行任务，直接返回
	rowData, ok := data.(map[string]interface{})
	if !ok {
		return nil, errors.New("invalid data type for SplitTask")
	}
	return rowData, nil
}

// ExampleTask 进行某种计算或数据转换
func ExampleTask(ctx context.Context, data interface{}) (interface{}, error) {
	// 期望 data 是 map[string]interface{}
	rowData, ok := data.(map[string]interface{})
	if !ok {
		return nil, errors.New("invalid data type for ExampleTask")
	}

	// 进行某种处理，例如添加一个字段
	rowData["Processed"] = true

	// 返回处理后的数据
	return rowData, nil
}

func main() {
	// 模拟从数据库读取10000行数据，分为5个批次，每批2000行
	totalRows := 10000
	batchSize := 2000
	batchCount := totalRows / batchSize

	// 准备批量输入数据
	inputData := make([]interface{}, batchCount)
	for batchID := 0; batchID < batchCount; batchID++ {
		batchData := make([]map[string]interface{}, batchSize)
		for i := 0; i < batchSize; i++ {
			globalRowIndex := batchID*batchSize + i
			batchData[i] = map[string]interface{}{
				"A": fmt.Sprintf("Row%d-ColA", globalRowIndex),
				"B": fmt.Sprintf("Row%d-ColB", globalRowIndex),
				"C": fmt.Sprintf("Row%d-ColC", globalRowIndex),
				"D": globalRowIndex,
				"E": globalRowIndex * 2,
			}
		}
		inputData[batchID] = batchData
	}

	// 创建 ExcelWriter
	excelWriter, err := NewExcelWriter("Sheet1")
	if err != nil {
		fmt.Printf("Failed to create Excel writer: %v\n", err)
		return
	}

	// 创建流水线
	pipeline := NewPipeline()

	// 添加流水线层
	pipeline.AddLayer(SplitTask, 10, 2*time.Second, 3)
	pipeline.AddLayer(ExampleTask, 20, 2*time.Second, 3)
	pipeline.AddLayer(ExampleTask, 20, 2*time.Second, 3)

	// 定义 WaitGroup，等待所有行被写入
	var writeWg sync.WaitGroup
	writeWg.Add(totalRows)

	// 定义回调函数，用于将结果写入 Excel
	callback := func(task *Task) {
		if task.Result == nil {
			fmt.Printf("Task GlobalRowIndex %d failed during processing.\n", task.GlobalRowIndex)
			writeWg.Done()
			return
		}
		// 将结果写入 Excel
		if err := excelWriter.WriteRow(task.ID, task.GlobalRowIndex, task.Result); err != nil {
			fmt.Printf("Failed to write Task GlobalRowIndex %d to Excel: %v\n", task.GlobalRowIndex, err)
			writeWg.Done()
		}
		writeWg.Done()
	}

	// 运行流水线
	err = pipeline.Run(inputData, callback)
	if err != nil {
		fmt.Printf("Pipeline error: %v\n", err)
	}

	// 等待所有行写入完成
	writeWg.Wait()

	// 保存并关闭 Excel 文件
	if err := excelWriter.Save("output.xlsx"); err != nil {
		fmt.Printf("Failed to save Excel file: %v\n", err)
	} else {
		fmt.Println("Excel file saved successfully.")
	}
}