io.Pipe() 导致 WaitGroup 卡住

从现在开始，我们要努力学习啦！今天我给大家带来《io.Pipe() 导致 WaitGroup 卡住》，感兴趣的朋友请继续看下去吧！下文中的内容我们主要会涉及到等等知识点，如果在阅读本文过程中有遇到不清楚的地方，欢迎留言呀！我们一起讨论，一起学习！

我正在处理一个巨大的数据文件，大约是。 100 gb。这个巨大文件中的每一行都是一段 json 数据，我想读取、压缩并将其存储在内存数据库中。

var wg sync.WaitGroup
for {
    line, err := reader.ReadString('\n')
    if err != nil {
        break
    }
    go func(index int) {
        wg.Add(1)
        pr, pw := io.Pipe()
        zw := lzw.NewWriter(pw, lzw.LSB, 8)
        _, err := io.Copy(zw, strings.NewReader(line))
        pw.Close()
        zw.Close()
        if err != nil {
            fmt.Println(err.Error())
        }
        b, err := io.ReadAll(pr)
        if err != nil {
            fmt.Println(err.Error())
        }
        client.Set(ctx, fmt.Sprintf("%d", index), base64.StdEncoding.EncodeToString(b), time.Hour*1000)
        pr.Close()
        wg.Done()
    }(index)
    if index%10000 == 0 {
        fmt.Println(index)
        wg.Wait()
    }
    index += 1
}

但是，此代码在处理前 10000 行后停止。当我在 zw.close() 之后向下移动 wg.add(1) 时，它会继续处理该行的其余部分（但变得不稳定）。当我尝试以未压缩的方式存储精确值时，如果没有 lzw 和 io.pipe() ，那么一切都可以正常工作。

我不确定我是否没有正确使用 waitgroup 或者是否有与 io.pipe() 相关的东西，而我还不知道。

正确答案

tldr：
1- 删除 pr, pw := io.pipe() 会使代码变得简单，因为它是多余，
尝试this：

line, err := reader.readstring('\n')
if err == io.eof {
    wg.wait()
    break
}
if err != nil {
    log.fatal(err)
}
wg.add(1)
go func(index int) {
    var buf bytes.buffer
    { // lexical scoping (static scoping)
        zw := lzw.newwriter(&buf, lzw.lsb, 8)
        n, err := zw.write([]byte(line)) // n, err := io.copy(zw, strings.newreader(line))
        if err != nil {
            log.fatal(err)
        }
        if int(n) != len(line) {
            log.fatal(n, len(line))
        }
        // it is the caller's responsibility to call close on the writecloser when finished writing.
        if err = zw.close(); err != nil {
            log.fatal(err)
        }
    }
    ctx, cancelfunc := context.withtimeout(context.background(), 100*time.millisecond)
    client.set(ctx, fmt.sprintf("%d", index), base64.stdencoding.encodetostring(buf.bytes()), 1000*time.hour)

    cancelfunc()
    wg.done()
}(index)

if index%tenthousand == 0 {
    wg.wait()
}

2- 您需要将 wg.add(1) 放在 go func(index int) {:

wg.add(1)
    go func(index int) {

3- wg.wait() 逻辑：

if index%10000 == 0 {
        fmt.println(index)
        wg.wait()
    }

如果 index%10000 != 0，最后一次迭代会发生什么。 因此，当 err == io.eof 时，您需要 wg.wait() 以使所有 goroutine 加入：

if err == io.eof {
    wg.wait()
    fmt.println("\n**** all done **** index =", index)
    break
}

4-您可以使用词法作用域（静态作用域）来限制某些变量作用域并使代码更易于管理 - 并了解何时 clos​​e lzw.newwriter ：

{ // lexical scoping (static scoping)
    zw := lzw.newwriter(bufio.newwriter(&buf), lzw.lsb, 8)
    n, err := io.copy(zw, strings.newreader(line))
    if err != nil {
        log.fatal(err)
    }
    if int(n) != len(line) {
        log.fatal(n, len(line))
    }
    // it is the caller's responsibility to call close on the writecloser when finished writing.
    if err = zw.close(); err != nil {
        log.fatal(err)
    }
}

5-始终检查错误，例如：

if err = zw.close(); err != nil {
    log.fatal(err)
}

这是接近您代码的工作版本 - 尝试 this 只是为了试验并发逻辑，看看会发生什么（不推荐，因为它有多余的 goroutine 和 io.pipe - 只是工作： p>

package main

import (
    "bufio"
    "compress/lzw"
    "context"
    "encoding/base64"
    "fmt"
    "io"
    "log"
    "strings"
    "sync"
    "time"
)

func main() {
    index := 0
    client := &myclient{}
    reader := bufio.newreader(file)
    // your code:
    var wg sync.waitgroup
    for {
        index++
        line, err := reader.readstring('\n')
        if err != nil {
            msg <- fmt.sprint(index, " done not waiting with err: ", err, time.now())
            wg.wait() // break waiting // if index%tenthousand != 0
            break
        }
        wg.add(1)
        go func(i int) {
            msg <- fmt.sprint(i, " enter running ... ", time.now())
            asyncreader, asyncwriter := io.pipe() // make it async to read and write
            zipwriter := lzw.newwriter(asyncwriter, lzw.lsb, 8)
            go func() { // async
                _, err := io.copy(zipwriter, strings.newreader(line))
                if err != nil {
                    log.fatal(err)
                }
                _ = zipwriter.close()
                _ = asyncwriter.close() // for io.readall
            }()
            b, err := io.readall(asyncreader)
            if err != nil {
                log.fatal(err)
            }
            client.set(context.background(), fmt.sprintf("%d", i), base64.stdencoding.encodetostring(b), time.hour*1000)
            asyncreader.close()
            time.sleep(1 * time.second)
            msg <- fmt.sprint(i, " exit running ... ", time.now())
            wg.done()
        }(index)

        msg <- fmt.sprint(index, " ", index%tenthousand == 0, " after go call")
        if index%tenthousand == 0 {
            wg.wait()
            msg <- fmt.sprint("..", index, " done waiting after go call. ", time.now())
        }
    }
    msg <- "bye forever."

    wg.wait()
    close(msg)
    wgmsg.wait()
}

// just for the go playground:
const tenthousand = 2

type myclient struct {
}

func (p *myclient) set(ctx context.context, a, b string, t time.duration) {
    // fmt.println("a =", a, ", b =", b, ", t =", t)
    if ctx.err() != nil {
        fmt.println(ctx.err())
    }
}

var file, myw = io.pipe()

func init() {
    go func() {
        for i := 1; i <= tenthousand+1; i++ {
            fmt.fprintf(myw, "%d text to compress aaaaaaaaaaaaaa\n", i)
        }
        myw.close()
    }()
    wgmsg.add(1)
    go func() {
        defer wgmsg.done()
        for s := range msg {
            fmt.println(s)
        }
    }()
}

var msg = make(chan string, 100)
var wgmsg sync.waitgroup

输出：

1 false after go call
2 true after go call
1 Enter running ... 2009-11-10 23:00:00 +0000 UTC m=+0.000000001
2 Enter running ... 2009-11-10 23:00:00 +0000 UTC m=+0.000000001
1 Exit running ... 2009-11-10 23:00:01 +0000 UTC m=+1.000000001
2 Exit running ... 2009-11-10 23:00:01 +0000 UTC m=+1.000000001
..2 Done waiting after go call. 2009-11-10 23:00:01 +0000 UTC m=+1.000000001
3 false after go call
3 Enter running ... 2009-11-10 23:00:01 +0000 UTC m=+1.000000001
4 Done not waiting with err: EOF 2009-11-10 23:00:01 +0000 UTC m=+1.000000001
3 Exit running ... 2009-11-10 23:00:02 +0000 UTC m=+2.000000001
Bye forever.

好了，本文到此结束，带大家了解了《io.Pipe() 导致 WaitGroup 卡住》，希望本文对你有所帮助！关注golang学习网公众号，给大家分享更多Golang知识！