如何使用Goroutine批量處理檔案？-有解無憂

假設我有一堆檔案要處理（比如 1000 個或更多），首先它們應該由 function 處理A()，functionA()將生成一個檔案，然后這個檔案將由 B() 處理。

如果我們一個一個地做，那太慢了，所以我想用 goroutine 一次處理 5 個檔案（我們一次不能處理太多，因為 CPU 無法承受）。

我是golang的新手，不確定我的想法是否正確，我認為函式A()是生產者，函式B()是消費者，函式B()會處理函式產生的檔案，A()我在下面寫了一些代碼，見諒，我真的不知道怎么寫代碼，誰能幫幫我？先感謝您！

package main

import "fmt"

var Box = make(chan string, 1024)

func A(file string) {
    fmt.Println(file, "is processing in func A()...")
    fileGenByA := "/path/to/fileGenByA1"
    Box <- fileGenByA
}

func B(file string) {
    fmt.Println(file, "is processing in func B()...")
}

func main() {
    // assuming that this is the file list read from a directory
    fileList := []string{
        "/path/to/file1",
        "/path/to/file2",
        "/path/to/file3",
    }

    // it seems I can't do this, because fileList may have 1000 or more file
    for _, v := range fileList {
        go A(v)
    }

    // can I do this?
    for file := range Box {
        go B(file)
    }
}

更新：

對不起，也許我沒有說清楚，實際上由函式生成的檔案A()存盤在硬碟中，而不是在變數（記憶體）中，所以它不必B()立即傳遞給函式。

uj5u.com熱心網友回復：

你已經成功了一半。您需要解決一些問題：

你的程式死鎖是因為沒有關閉任何東西Box，所以 main 函式永遠無法完成range它。
你不是在等待你的 goroutines 完成，而且有超過 5 個 goroutines。（這些的解決方案太交織在一起，無法單獨描述）

1. 死鎖

fatal error: all goroutines are asleep - deadlock!

goroutine 1 [chan receive]:
main.main()

當您range通過通道時，您會從通道中讀取每個值，直到它同時為 closed 和 empty。既然你從來沒有close這個頻道，那range這個頻道就永遠無法完成，程式也永遠無法完成。

在您的情況下，這是一個相當容易解決的問題：我們只需要在知道不會有更多寫入通道時關閉通道。

    for _, v := range fileList {
        go A(v)
    }
    close(Box)

請記住，close讀取通道并不會阻止它被讀取，只會阻止它被寫入。現在消費者可以區分未來可能接收更多資料的空通道和永遠不會接收更多資料的空通道。

一旦你添加了close(Box)，程式就不會再死鎖了，但它仍然不起作用。

2. 太多的 Goroutines 而沒有等待它們完成

要運行某個最大數量的并發執行，而不是為每個輸入創建一個 goroutine，而是在“作業池”中創建 goroutine：

創建一個通道來傳遞工人的作業
為 goroutines 創建一個通道以回傳其結果（如果有）
啟動你想要的 goroutines 數量
至少啟動一個額外的 goroutine 來分派作業或收集結果，因此您不必嘗試從主 goroutine 執行這兩項操作
使用 async.WaitGroup等待所有資料被處理
close通道向作業人員和結果收集器發出信號，表明他們的通道已完成填充。

在我們進入實作之前，讓我們談談如何A和B互動。

首先它們應該由函式A（）處理，函式A（）會生成一個檔案，然后這個檔案將由B（）處理。

A() and B() must, then, execute serially. They can still pass their data through a channel, but since their execution must be serial, it does nothing for you. Simpler is to run them sequentially in the workers. For that, we'll need to change A() to either call B, or to return the path for B and the worker can call. I choose the latter.

func A(file string) string {
    fmt.Println(file, "is processing in func A()...")
    fileGenByA := "/path/to/fileGenByA1"
    return fileGenByA
}

Before we write our worker function, we also must consider the result of B. Currently, B returns nothing. In the real world, unless B() cannot fail, you would at least want to either return the error, or at least panic. I'll skip over collecting results for now.

Now we can write our worker function.

func worker(wg *sync.WaitGroup, incoming <-chan string) {
    defer wg.Done()
    for file := range incoming {
        B(A(file))
    }
}

Now all we have to do is start 5 such workers, write the incoming files to the channel, close it, and wg.Wait() for the workers to complete.

    incoming_work := make(chan string)
    var wg sync.WaitGroup
    for i := 0; i < 5; i   {
        wg.Add(1)
        go worker(&wg, incoming_work)
    }
    for _, v := range fileList {
        incoming_work <- v
    }
    close(incoming_work)
    wg.Wait()

Full example at https://go.dev/play/p/A1H4ArD2LD8

Returning Results.

It's all well and good to be able to kick off goroutines and wait for them to complete. But what if you need results back from your goroutines? In all but the simplest of cases, you would at least want to know if files failed to process so you could investigate the errors.

We have only 5 workers, but we have many files, so we have many results. Each worker will have to return several results. So, another channel. It's usually worth defining a struct for your return:

type result struct {
  file string
  err error
}

This tells us not just whether there was an error but also clearly defines which file from which the error resulted.

How will we test an error case in our current code? In your example, B always gets the same value from A. If we add A's incoming file name to the path it passes to B, we can mock an error based on a substring. My mocked error will be that file3 fails.

func A(file string) string {
    fmt.Println(file, "is processing in func A()...")
    fileGenByA := "/path/to/fileGenByA1/"   file
    return fileGenByA
}

func B(file string) (r result) {
    r.file = file
    fmt.Println(file, "is processing in func B()...")
    if strings.Contains(file, "file3") {
        r.err = fmt.Errorf("Test error")
    }
    return
}

Our workers will be sending results, but we need to collect them somewhere. main() is busy dispatching work to the workers, blocking on its write to incoming_work when the workers are all busy. So the simplest place to collect the results is another goroutine. Our results collector goroutine has to read from a results channel, print out errors for debugging, and the return the total number of failures so our program can return a final exit status indicating overall success or failure.

    failures_chan := make(chan int)
    go func() {
        var failures int
        for result := range results {
            if result.err != nil {
                failures  
                fmt.Printf("File %s failed: %s", result.file, result.err.Error())
            }
        }
        failures_chan <- failures

    }()

Now we have another channel to close, and it's important we close it after all workers are done. So we close(results) after we wg.Wait() for the workers.

    close(incoming_work)
    wg.Wait()
    close(results)
    if failures := <-failures_chan; failures > 0 {
        os.Exit(1)
    }

Putting all that together, we end up with this code:

package main

import (
    "fmt"
    "os"
    "strings"
    "sync"
)

func A(file string) string {
    fmt.Println(file, "is processing in func A()...")
    fileGenByA := "/path/to/fileGenByA1/"   file
    return fileGenByA
}

func B(file string) (r result) {
    r.file = file
    fmt.Println(file, "is processing in func B()...")
    if strings.Contains(file, "file3") {
        r.err = fmt.Errorf("Test error")
    }
    return
}

func worker(wg *sync.WaitGroup, incoming <-chan string, results chan<- result) {
    defer wg.Done()
    for file := range incoming {
        results <- B(A(file))
    }
}

type result struct {
    file string
    err  error
}

func main() {
    // assuming that this is the file list read from a directory
    fileList := []string{
        "/path/to/file1",
        "/path/to/file2",
        "/path/to/file3",
    }
    incoming_work := make(chan string)
    results := make(chan result)
    var wg sync.WaitGroup
    for i := 0; i < 5; i   {
        wg.Add(1)
        go worker(&wg, incoming_work, results)
    }
    failures_chan := make(chan int)
    go func() {
        var failures int
        for result := range results {
            if result.err != nil {
                failures  
                fmt.Printf("File %s failed: %s", result.file, result.err.Error())
            }
        }
        failures_chan <- failures

    }()
    for _, v := range fileList {
        incoming_work <- v
    }
    close(incoming_work)
    wg.Wait()
    close(results)
    if failures := <-failures_chan; failures > 0 {
        os.Exit(1)
    }
}

And when we run it, we get:

/path/to/file1 is processing in func A()...
/path/to/fileGenByA1//path/to/file1 is processing in func B()...
/path/to/file2 is processing in func A()...
/path/to/fileGenByA1//path/to/file2 is processing in func B()...
/path/to/file3 is processing in func A()...
/path/to/fileGenByA1//path/to/file3 is processing in func B()...
File /path/to/fileGenByA1//path/to/file3 failed: Test error
Program exited.

A final thought: buffered channels.

There is nothing wrong with buffered channels. Especially if you know the overall size of incoming work and results, buffered channels can obviate the results collector goroutine because you can allocate a buffered channel big enough to hold all results. However, I think it's more straightforward to understand this pattern if the channels are unbuffered. The key takeaway is that you don't need to know the number of incoming or outgoing results, which could indeed be different numbers or based on something that can't be predetermined.

uj5u.com熱心網友回復：

您可以生成 5 個從作業通道讀取的 goroutine。這樣你就可以一直運行 5 個 goroutine 并且不需要對它們進行批處理，這樣你就必須等到 5 個完成才能開始下一個 5 個。

func main() {
    stack := []string{
        "foo",
        "bar",
        "baz",
        "qux",
        "quux",
        "corge",
    }

    work := make(chan string)
    results := make(chan string)

    // create 5 go routines
    wg := sync.WaitGroup{}
    for i := 0; i < 5; i   {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for s := range work {
                results <- B(A(s))
            }
        }()
    }

    // collect the results
    go func() {
        for result := range results {
            fmt.Println(result)
        }
    }()

    // send the work to the workers
    for _, s := range stack {
        work <- s
    }
    close(work)

    // wait for the workers to finish
    // then close the results channel
    wg.Wait()
    close(results)
}

https://play.golang.com/p/IgoMfAR-Tya

uj5u.com熱心網友回復：

請檢查這個。

package main

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

var batchSize = 5

func A(file string, releaseReq chan struct{}, box chan string, done *sync.WaitGroup) {
    defer func() {
        <-releaseReq
        done.Done()
    }()
    time.Sleep(2 * time.Second)
    fmt.Println(file, "is processing in func A()...")
    fileGenByA := "/path/to/fileGenByA1"
    box <- fileGenByA
}

func B(file string, done *sync.WaitGroup) {
    defer func() {
        done.Done()
    }()
    time.Sleep(1 * time.Second)
    fmt.Println(file, "is processing in func B()...")
}

func main() {
    fileList := []string{
        "/path/to/file1",
        "/path/to/file2",
        "/path/to/file3",
        "/path/to/file4",
        "/path/to/file5",
        "/path/to/file6",
        "/path/to/file7",
        "/path/to/file8",
        "/path/to/file9",
        "/path/to/file10",
    }
    box := make(chan string, 5)

    var doneProcessA sync.WaitGroup
    doneProcessA.Add(1)
    go func() {
        rateLimitter := make(chan struct{}, 5)
        var processA sync.WaitGroup
        for _, v := range fileList {
            rateLimitter <- struct{}{}
            processA.Add(1)
            go A(v, rateLimitter, box, &processA)
        }
        processA.Wait()
        doneProcessA.Done()
        close(box)
    }()

    var doneProcessB sync.WaitGroup
    doneProcessB.Add(1)
    go func() {
        var processB sync.WaitGroup
        for file := range box {
            processB.Add(1)
            go B(file, &processB)
        }
        processB.Wait()
        doneProcessB.Done()
    }()
    doneProcessA.Wait()
    doneProcessB.Wait()
}

轉載請註明出處，本文鏈接：https://www.uj5u.com/caozuo/442995.html

標籤：走协程

上一篇：restTemplate.postForEntity的單元測驗導致ResourceAccessException

下一篇：如何將中間件用于特定路由等？