正确监测长时间运行任务的进展

对于一个Golang开发者来说，牢固扎实的基础是十分重要的，golang学习网就来带大家一点点的掌握基础知识点。今天本篇文章带大家了解《正确监测长时间运行任务的进展》，主要介绍了，希望对大家的知识积累有所帮助，快点收藏起来吧，否则需要时就找不到了！

我想跟踪一些长时间运行的进程的执行情况并显示用户完成百分比和错误（如果有）。如果它是一个长时间运行的流程，那么很简单 - 您可以创建进度（百分比）和错误的通道。当我们有 x 个长时间运行的进程时，实现此类逻辑的正确方法是什么？

下面是一段有效的代码片段，但我不太喜欢它的实现方式。 我创建了一个结构 progresstracker ，它保留 url （作为字段）、error、progress 作为渠道。我将这样的 progresstracker 保留在一个切片中，一旦提交所有任务，我就会通过 progresstracker 的切片进行迭代，并监听 progresstracker 中每个 tracker 的通道。一旦提交的请求数 == 接收到的响应数 - 退出循环。

这是 go 惯用的解决方案吗？将 progresstracker 作为通道传递给函数会更容易，但我不知道在这种情况下如何正确发送“进度”、“错误”和“完成”事件。

代码如下，go演示中也有同样的代码：https://go.dev/play/p/f3hxjszr9wv

package main

import (
    "errors"
    "fmt"
    "strings"
    "sync"
    "time"
)

type progresstracker struct {
    progress chan int
    error    chan error
    complete chan bool
    url      string
}

/**
this method sleeps for 1 second and sends progress (in %) in each iteration to progress channel
for .net sites on 3rd iteration fail with error
when everything is completed, send a message to complete channel
*/
func work(url string, tracker *progresstracker) {
    tracker.url = url
    fmt.printf("processing url %s\n", url)
    for i := 1; i <= 5; i++ {
        time.sleep(time.second)
        if i == 3 && strings.hassuffix(url, ".net") {
            tracker.error <- errors.new("emulating error for .net sites")
            tracker.complete <- true
        }
        progress := 20 * i
        tracker.progress <- progress
    }
    tracker.complete <- true
}

func main() {
    var trackers []*progresstracker
    var urls = []string{"google.com", "youtube.com", "someurl.net"}
    var wg sync.waitgroup
    
    wg.add(len(urls))
    for _, url := range urls {
        tracker := &progresstracker{
            progress: make(chan int),
            error:    make(chan error),
            complete: make(chan bool),
        }
        trackers = append(trackers, tracker)
        go func(workurl string, progresstracker *progresstracker) {
            work(workurl, progresstracker)
        }(url, tracker)
    }

    go func() {
        wg.wait()
    }()

    var processed = 0

    //iterate through all trackers and select each channel.
    //exit from this loop when number of processed requests equals the number of trackers
    for {
        for _, t := range trackers {
            select {
            case pr := <-t.progress:
                fmt.printf("url = %s, progress = %d\n", t.url, pr)
            case err := <-t.error:
                fmt.printf("url = %s, error = %s\n", t.url, err.error())
            case <-t.complete:
                fmt.printf("url = %s is completed\n", t.url)
                processed = processed + 1
                if processed == len(trackers) {
                    fmt.printf("everything is completed, exit")
                    return
                }
            }
        }
    }
}

更新： 如果我为其中一个任务添加延迟，那么我选择所有通道的 for 循环也将等待每次迭代中最慢的工作线程。 go 演示：https://go.dev/play/p/9fvde7zgirp 更新的工作功能：

func work(url string, tracker *ProgressTracker) {
    tracker.Url = url
    fmt.Printf("processing url %s\n", url)
    for i := 1; i <= 5; i++ {
        if url == "google.com" {
            time.Sleep(time.Second * 3)
        }
        time.Sleep(time.Second)
        if i == 3 && strings.HasSuffix(url, ".net") {
            tracker.Error <- errors.New("emulating error for .net sites")
            tracker.Complete <- true
            return
        }
        progress := 20 * i
        tracker.Progress <- progress
    }
    tracker.Complete <- true
}

正确答案

谢谢您的回答！我想出了这种方法，它适合我的需求：

package main

import (
    "errors"
    "fmt"
    "strings"
    "sync"
    "time"
)

type ProgressTracker struct {
    Progress  int
    Error     error
    Completed bool
    Url       string
}

/**
This method sleeps for 1 second and sends progress (in %) in each iteration to Progress channel
For .net sites on 3rd iteration fail with error
When everything is completed, send a message to Complete channel
*/
func work(url string, tracker chan ProgressTracker) {
    var internalTracker = ProgressTracker{
        Url: url,
    }
    tracker <- internalTracker
    fmt.Printf("processing url %s\n", url)
    for i := 1; i <= 5; i++ {
        if url == "google.com" {
            time.Sleep(time.Second * 3)
        }
        time.Sleep(time.Second)
        if i == 3 && strings.HasSuffix(url, ".net") {
            internalTracker.Error = errors.New("error for .net sites")
            internalTracker.Completed = true
            tracker <- internalTracker
            return
        }
        progress := 20 * i
        internalTracker.Progress = progress
        internalTracker.Completed = false
        tracker <- internalTracker
    }
    internalTracker.Completed = true
    tracker <- internalTracker
}

func main() {
    var urls = []string{"google.com", "youtube.com", "someurl.net"}
    var tracker = make(chan ProgressTracker, len(urls))
    var wg sync.WaitGroup
    wg.Add(len(urls))

    for _, url := range urls {
        go func(workUrl string) {
            defer wg.Done()
            work(workUrl, tracker)
        }(url)
    }

    go func() {
        wg.Wait()
        close(tracker)
        fmt.Printf("After wg wait")
    }()

    var completed = 0

    for completed < len(urls) {
        select {
        case t := <-tracker:
            if t.Completed {
                fmt.Printf("Processing for %s is completed!\n", t.Url)
                completed = completed + 1
            } else {
                fmt.Printf("Processing for %s is in progress: %d\n", t.Url, t.Progress)
            }
            if t.Error != nil {
                fmt.Printf("Url %s has errors %s\n", t.Url, t.Error)
            }
        }

    }
}

在这里，我将 progresstracker 作为通道传递（progresstracker 中的字段被声明为简单字段，而不是通道），并且在工作函数中的每个事件上返回正在发生的情况的完整状态（如果进度增加 - 设置新值并返回一个结构 到通道，如果发生错误 - 设置错误并返回结构等）。

您陷入僵局，因为您继续对已完成但没有默认值的跟踪器进行 select 。您的内部 for 循环每次都会迭代所有跟踪器，其中包括已完成并且永远不会发送其他消息的跟踪器。解决这个问题的最简单的方法是一个空的 default ，这也将使这些在现实生活中表现得更好，因为它们并不都以相同的速度进行，但它确实会将其变成一个紧密的循环，从而消耗更多的 cpu。 p>

您的 waitgroup 根本不执行任何操作；你在 goroutine 中调用 wait 但当它返回时什么也不做，并且你从未在它跟踪的 goroutine 中调用 done，所以它永远不会返回。相反，您单独跟踪收到的 complete 消息的数量，并使用它而不是 waitgroup；目前还不清楚为什么要这样实现。

修复这两个问题可以解决所述问题：https://go.dev/play/p/do0g9jrX0mY

但是，这可能不是正确的方法。不可能用一个人为的例子来说明什么是正确的方法；如果示例是它需要做的全部，那么您不需要任何逻辑，您可以将打印语句放在工作人员中，只需使用等待组而不使用通道即可完成。假设您实际上正在对结果执行某些操作，您可能需要一个由所有工作人员共享的单个 completed 通道和一个 error 通道，并且可能需要一个完全不同的机制来跟踪进度，例如您可以从中读取的原子 int/float当您想了解当前进度时。那么您就不需要嵌套循环的东西，您只需一个带有 select 的循环来从共享通道读取消息。这完全取决于此代码的使用上下文。

本篇关于《正确监测长时间运行任务的进展》的介绍就到此结束啦，但是学无止境，想要了解学习更多关于Golang的相关知识，请关注golang学习网公众号！