Golang中如何追踪任务序列的执行进度

在Go中实现任务序列进度追踪可以通过构建一个进度管理器来处理。以下是一个完整的解决方案示例：

package main

import (
    "fmt"
    "sync"
    "sync/atomic"
    "time"
)

// ProgressReporter 定义进度报告接口
type ProgressReporter interface {
    ReportProgress(progress float64)
}

// TaskProgress 表示单个任务的进度
type TaskProgress struct {
    ID         string
    Weight     float64  // 任务权重
    Progress   float64  // 0-1之间的进度
    SubTasks   []*TaskProgress
    IsParallel bool
    mu         sync.RWMutex
}

// ProgressManager 管理整体进度
type ProgressManager struct {
    tasks      []*TaskProgress
    totalWeight float64
    listener   func(overallProgress float64)
    mu         sync.RWMutex
}

// NewProgressManager 创建新的进度管理器
func NewProgressManager() *ProgressManager {
    return &ProgressManager{
        tasks:      make([]*TaskProgress, 0),
        totalWeight: 0,
    }
}

// AddTask 添加任务到进度管理器
func (pm *ProgressManager) AddTask(id string, weight float64, isParallel bool) *TaskProgress {
    pm.mu.Lock()
    defer pm.mu.Unlock()
    
    task := &TaskProgress{
        ID:         id,
        Weight:     weight,
        Progress:   0,
        SubTasks:   make([]*TaskProgress, 0),
        IsParallel: isParallel,
    }
    
    pm.tasks = append(pm.tasks, task)
    pm.totalWeight += weight
    
    return task
}

// AddSubTask 为任务添加子任务
func (tp *TaskProgress) AddSubTask(id string, weight float64) *TaskProgress {
    tp.mu.Lock()
    defer tp.mu.Unlock()
    
    subTask := &TaskProgress{
        ID:       id,
        Weight:   weight,
        Progress: 0,
        SubTasks: make([]*TaskProgress, 0),
    }
    
    tp.SubTasks = append(tp.SubTasks, subTask)
    return subTask
}

// ReportProgress 报告任务进度
func (tp *TaskProgress) ReportProgress(progress float64) {
    tp.mu.Lock()
    defer tp.mu.Unlock()
    
    if progress < 0 {
        progress = 0
    }
    if progress > 1 {
        progress = 1
    }
    
    tp.Progress = progress
}

// CalculateOverallProgress 计算整体进度
func (pm *ProgressManager) CalculateOverallProgress() float64 {
    pm.mu.RLock()
    defer pm.mu.RUnlock()
    
    if pm.totalWeight == 0 {
        return 0
    }
    
    var totalProgress float64
    for _, task := range pm.tasks {
        taskProgress := task.calculateTaskProgress()
        totalProgress += taskProgress * task.Weight
    }
    
    return totalProgress / pm.totalWeight
}

// calculateTaskProgress 计算单个任务的进度（考虑子任务）
func (tp *TaskProgress) calculateTaskProgress() float64 {
    tp.mu.RLock()
    defer tp.mu.RUnlock()
    
    if len(tp.SubTasks) == 0 {
        return tp.Progress
    }
    
    if tp.IsParallel {
        // 并行子任务：取平均进度
        var totalProgress float64
        var totalWeight float64
        
        for _, subTask := range tp.SubTasks {
            subTaskProgress := subTask.calculateTaskProgress()
            totalProgress += subTaskProgress * subTask.Weight
            totalWeight += subTask.Weight
        }
        
        if totalWeight == 0 {
            return 0
        }
        return totalProgress / totalWeight
    } else {
        // 串行子任务：取最小进度（最慢的任务决定整体）
        minProgress := 1.0
        for _, subTask := range tp.SubTasks {
            subTaskProgress := subTask.calculateTaskProgress()
            if subTaskProgress < minProgress {
                minProgress = subTaskProgress
            }
        }
        return minProgress
    }
}

// SetProgressListener 设置进度监听器
func (pm *ProgressManager) SetProgressListener(listener func(overallProgress float64)) {
    pm.listener = listener
}

// NotifyListener 通知监听器（如果设置）
func (pm *ProgressManager) NotifyListener() {
    if pm.listener != nil {
        progress := pm.CalculateOverallProgress()
        pm.listener(progress)
    }
}

// 使用示例
func main() {
    pm := NewProgressManager()
    
    // 设置进度监听器
    pm.SetProgressListener(func(progress float64) {
        fmt.Printf("总体进度: %.2f%%\n", progress*100)
    })
    
    // 添加主任务
    task1 := pm.AddTask("data_processing", 0.4, false)
    task2 := pm.AddTask("file_upload", 0.3, true)
    task3 := pm.AddTask("validation", 0.3, false)
    
    // 为任务1添加串行子任务
    subTask1_1 := task1.AddSubTask("data_extraction", 0.4)
    subTask1_2 := task1.AddSubTask("data_transformation", 0.6)
    
    // 为任务2添加并行子任务
    subTask2_1 := task2.AddSubTask("upload_file1", 0.5)
    subTask2_2 := task2.AddSubTask("upload_file2", 0.5)
    
    // 模拟任务执行
    go func() {
        // 任务1的子任务
        for i := 0; i <= 10; i++ {
            subTask1_1.ReportProgress(float64(i) / 10)
            pm.NotifyListener()
            time.Sleep(100 * time.Millisecond)
        }
        
        for i := 0; i <= 10; i++ {
            subTask1_2.ReportProgress(float64(i) / 10)
            pm.NotifyListener()
            time.Sleep(150 * time.Millisecond)
        }
    }()
    
    go func() {
        // 任务2的并行子任务
        for i := 0; i <= 5; i++ {
            subTask2_1.ReportProgress(float64(i) / 5)
            subTask2_2.ReportProgress(float64(i) / 5)
            pm.NotifyListener()
            time.Sleep(200 * time.Millisecond)
        }
    }()
    
    go func() {
        // 任务3
        for i := 0; i <= 10; i++ {
            task3.ReportProgress(float64(i) / 10)
            pm.NotifyListener()
            time.Sleep(120 * time.Millisecond)
        }
    }()
    
    // 等待所有任务完成
    time.Sleep(3 * time.Second)
    finalProgress := pm.CalculateOverallProgress()
    fmt.Printf("最终进度: %.2f%%\n", finalProgress*100)
}

这个实现提供了：

1. 权重支持：每个任务可以设置不同的权重来反映其重要性或耗时
2. 并行/串行任务处理：并行任务取平均进度，串行任务取最慢进度
3. 嵌套任务支持：支持多级子任务
4. 线程安全：使用读写锁保护并发访问
5. 进度监听：可以设置回调函数实时获取进度更新

对于现成的解决方案，社区中有一些相关库：

• github.com/vbauerster/mpb - 多进度条库
• github.com/cheggaaa/pb - 进度条库
• github.com/schollz/progressbar - 简单进度条

但这些库主要专注于进度条显示，对于复杂的嵌套任务进度追踪，通常需要根据具体需求定制实现。