{"id":207039,"date":"2025-07-08T08:37:59","date_gmt":"2025-07-08T00:37:59","guid":{"rendered":"https:\/\/server.hk\/cnblog\/207039\/"},"modified":"2025-07-08T08:37:59","modified_gmt":"2025-07-08T00:37:59","slug":"io-pipe-%e5%af%bc%e8%87%b4-waitgroup-%e5%8d%a1%e4%bd%8f","status":"publish","type":"post","link":"https:\/\/server.hk\/cnblog\/207039\/","title":{"rendered":"io.Pipe() \u5bfc\u81f4 WaitGroup \u5361\u4f4f"},"content":{"rendered":"

<\/b> <\/p>\n

\u5f53\u524d\u4f4d\u7f6e\uff1a ><\/span> ><\/span> ><\/span> ><\/span> io.Pipe() \u5bfc\u81f4 WaitGroup \u5361\u4f4f<\/span><\/p>\n

\u6765\u6e90\uff1astackoverflow<\/span>
\n2024-04-19 17:00:35<\/span>
\n<\/i>0\u6d4f\u89c8<\/span>
\n<\/i>\u6536\u85cf<\/span> <\/p>\n

\u4ece\u73b0\u5728\u5f00\u59cb\uff0c\u6211\u4eec\u8981\u52aa\u529b\u5b66\u4e60\u5566\uff01\u4eca\u5929\u6211\u7ed9\u5927\u5bb6\u5e26\u6765\u300aio.Pipe() \u5bfc\u81f4 WaitGroup \u5361\u4f4f\u300b<\/span>\uff0c\u611f\u5174\u8da3\u7684\u670b\u53cb\u8bf7\u7ee7\u7eed\u770b\u4e0b\u53bb\u5427\uff01\u4e0b\u6587\u4e2d\u7684\u5185\u5bb9\u6211\u4eec\u4e3b\u8981\u4f1a\u6d89\u53ca\u5230<\/span>\u7b49\u7b49\u77e5\u8bc6\u70b9\uff0c\u5982\u679c\u5728\u9605\u8bfb\u672c\u6587\u8fc7\u7a0b\u4e2d\u6709\u9047\u5230\u4e0d\u6e05\u695a\u7684\u5730\u65b9\uff0c\u6b22\u8fce\u7559\u8a00\u5440\uff01\u6211\u4eec\u4e00\u8d77\u8ba8\u8bba\uff0c\u4e00\u8d77\u5b66\u4e60\uff01<\/p>\n

\u95ee\u9898\u5185\u5bb9
\n <\/p>\n

\u6211\u6b63\u5728\u5904\u7406\u4e00\u4e2a\u5de8\u5927\u7684\u6570\u636e\u6587\u4ef6\uff0c\u5927\u7ea6\u662f\u3002 100 gb\u3002\u8fd9\u4e2a\u5de8\u5927\u6587\u4ef6\u4e2d\u7684\u6bcf\u4e00\u884c\u90fd\u662f\u4e00\u6bb5 json \u6570\u636e\uff0c\u6211\u60f3\u8bfb\u53d6\u3001\u538b\u7f29\u5e76\u5c06\u5176\u5b58\u50a8\u5728\u5185\u5b58\u6570\u636e\u5e93\u4e2d\u3002<\/p>\n

var wg sync.WaitGroup\nfor {\n    line, err := reader.ReadString('\\n')\n    if err != nil {\n        break\n    }\n    go func(index int) {\n        wg.Add(1)\n        pr, pw := io.Pipe()\n        zw := lzw.NewWriter(pw, lzw.LSB, 8)\n        _, err := io.Copy(zw, strings.NewReader(line))\n        pw.Close()\n        zw.Close()\n        if err != nil {\n            fmt.Println(err.Error())\n        }\n        b, err := io.ReadAll(pr)\n        if err != nil {\n            fmt.Println(err.Error())\n        }\n        client.Set(ctx, fmt.Sprintf(\"%d\", index), base64.StdEncoding.EncodeToString(b), time.Hour*1000)\n        pr.Close()\n        wg.Done()\n    }(index)\n    if index%10000 == 0 {\n        fmt.Println(index)\n        wg.Wait()\n    }\n    index += 1\n}<\/pre>\n
\u4f46\u662f\uff0c\u6b64\u4ee3\u7801\u5728\u5904\u7406\u524d 10000 \u884c\u540e\u505c\u6b62\u3002\u5f53\u6211\u5728 zw.close()<\/code> \u4e4b\u540e\u5411\u4e0b\u79fb\u52a8 wg.add(1)<\/code> \u65f6\uff0c\u5b83\u4f1a\u7ee7\u7eed\u5904\u7406\u8be5\u884c\u7684\u5176\u4f59\u90e8\u5206\uff08\u4f46\u53d8\u5f97\u4e0d\u7a33\u5b9a\uff09\u3002\u5f53\u6211\u5c1d\u8bd5\u4ee5\u672a\u538b\u7f29\u7684\u65b9\u5f0f\u5b58\u50a8\u7cbe\u786e\u503c\u65f6\uff0c\u5982\u679c\u6ca1\u6709 lzw<\/code> \u548c io.pipe()<\/code> \uff0c\u90a3\u4e48\u4e00\u5207\u90fd\u53ef\u4ee5\u6b63\u5e38\u5de5\u4f5c\u3002<\/p>\n
\u6211\u4e0d\u786e\u5b9a\u6211\u662f\u5426\u6ca1\u6709\u6b63\u786e\u4f7f\u7528 waitgroup<\/code> \u6216\u8005\u662f\u5426\u6709\u4e0e io.pipe()<\/code> \u76f8\u5173\u7684\u4e1c\u897f\uff0c\u800c\u6211\u8fd8\u4e0d\u77e5\u9053\u3002<\/p>\n
 <\/p>\n
\u6b63\u786e\u7b54\u6848<\/h2>\n
 <\/p>\n
tldr\uff1a
 1- \u5220\u9664 pr, pw := io.pipe()<\/code> \u4f1a\u4f7f\u4ee3\u7801\u53d8\u5f97\u7b80\u5355<\/strong>\uff0c\u56e0\u4e3a\u5b83\u662f\u591a\u4f59<\/strong>\uff0c
 \u5c1d\u8bd5\uff1a<\/p>\n
line, err := reader.readstring('\\n')\nif err == io.eof {\n    wg.wait()\n    break\n}\nif err != nil {\n    log.fatal(err)\n}\nwg.add(1)\ngo func(index int) {\n    var buf bytes.buffer\n    { \/\/ lexical scoping (static scoping)\n        zw := lzw.newwriter(&buf, lzw.lsb, 8)\n        n, err := zw.write([]byte(line)) \/\/ n, err := io.copy(zw, strings.newreader(line))\n        if err != nil {\n            log.fatal(err)\n        }\n        if int(n) != len(line) {\n            log.fatal(n, len(line))\n        }\n        \/\/ it is the caller's responsibility to call close on the writecloser when finished writing.\n        if err = zw.close(); err != nil {\n            log.fatal(err)\n        }\n    }\n    ctx, cancelfunc := context.withtimeout(context.background(), 100*time.millisecond)\n    client.set(ctx, fmt.sprintf(\"%d\", index), base64.stdencoding.encodetostring(buf.bytes()), 1000*time.hour)\n\n    cancelfunc()\n    wg.done()\n}(index)\n\nif index%tenthousand == 0 {\n    wg.wait()\n}<\/pre>\n
2- \u60a8\u9700\u8981\u5c06 wg.add(1)<\/code> \u653e\u5728 go func(index int) {<\/code>:<\/p>\n
 \u4e4b\u524d <\/p>\n
wg.add(1)\n    go func(index int) {<\/pre>\n
3- wg.wait()<\/code> \u903b\u8f91\uff1a<\/p>\n
if index%10000 == 0 {\n        fmt.println(index)\n        wg.wait()\n    }<\/pre>\n
\u5982\u679c index%10000 != 0<\/code>\uff0c\u6700\u540e\u4e00\u6b21\u8fed\u4ee3\u4f1a\u53d1\u751f\u4ec0\u4e48\u3002 \u56e0\u6b64\uff0c\u5f53 err == io.eof<\/code> \u65f6\uff0c\u60a8\u9700\u8981 wg.wait()<\/code> \u4ee5\u4f7f\u6240\u6709 goroutine \u52a0\u5165\uff1a<\/p>\n
if err == io.eof {\n    wg.wait()\n    fmt.println(\"\\n**** all done **** index =\", index)\n    break\n}<\/pre>\n
4-\u60a8\u53ef\u4ee5\u4f7f\u7528\u8bcd\u6cd5\u4f5c\u7528\u57df<\/strong>\uff08\u9759\u6001\u4f5c\u7528\u57df\uff09\u6765\u9650\u5236\u67d0\u4e9b\u53d8\u91cf\u4f5c\u7528\u57df\u5e76\u4f7f\u4ee3\u7801\u66f4\u6613\u4e8e\u7ba1\u7406 – \u5e76\u4e86\u89e3\u4f55\u65f6 clos\u200b\u200be<\/code> lzw.newwriter<\/code> \uff1a<\/p>\n
{ \/\/ lexical scoping (static scoping)\n    zw := lzw.newwriter(bufio.newwriter(&buf), lzw.lsb, 8)\n    n, err := io.copy(zw, strings.newreader(line))\n    if err != nil {\n        log.fatal(err)\n    }\n    if int(n) != len(line) {\n        log.fatal(n, len(line))\n    }\n    \/\/ it is the caller's responsibility to call close on the writecloser when finished writing.\n    if err = zw.close(); err != nil {\n        log.fatal(err)\n    }\n}<\/pre>\n
5-\u59cb\u7ec8\u68c0\u67e5\u9519\u8bef\uff0c\u4f8b\u5982\uff1a<\/p>\n
if err = zw.close(); err != nil {\n    log.fatal(err)\n}<\/pre>\n
\u8fd9\u662f\u63a5\u8fd1\u60a8\u4ee3\u7801\u7684\u5de5\u4f5c\u7248\u672c – \u5c1d\u8bd5 \u53ea\u662f\u4e3a\u4e86\u8bd5\u9a8c\u5e76\u53d1\u903b\u8f91\uff0c\u770b\u770b\u4f1a\u53d1\u751f\u4ec0\u4e48\uff08\u4e0d\u63a8\u8350<\/strong>\uff0c\u56e0\u4e3a\u5b83\u6709\u591a\u4f59\u7684 goroutine \u548c io.pipe<\/code> – \u53ea\u662f\u5de5\u4f5c\uff1a p> <\/p>\n
package main\n\nimport (\n    \"bufio\"\n    \"compress\/lzw\"\n    \"context\"\n    \"encoding\/base64\"\n    \"fmt\"\n    \"io\"\n    \"log\"\n    \"strings\"\n    \"sync\"\n    \"time\"\n)\n\nfunc main() {\n    index := 0\n    client := &myclient{}\n    reader := bufio.newreader(file)\n    \/\/ your code:\n    var wg sync.waitgroup\n    for {\n        index++\n        line, err := reader.readstring('\\n')\n        if err != nil {\n            msg <- fmt.sprint(index, \" done not waiting with err: \", err, time.now())\n            wg.wait() \/\/ break waiting \/\/ if index%tenthousand != 0\n            break\n        }\n        wg.add(1)\n        go func(i int) {\n            msg <- fmt.sprint(i, \" enter running ... \", time.now())\n            asyncreader, asyncwriter := io.pipe() \/\/ make it async to read and write\n            zipwriter := lzw.newwriter(asyncwriter, lzw.lsb, 8)\n            go func() { \/\/ async\n                _, err := io.copy(zipwriter, strings.newreader(line))\n                if err != nil {\n                    log.fatal(err)\n                }\n                _ = zipwriter.close()\n                _ = asyncwriter.close() \/\/ for io.readall\n            }()\n            b, err := io.readall(asyncreader)\n            if err != nil {\n                log.fatal(err)\n            }\n            client.set(context.background(), fmt.sprintf(\"%d\", i), base64.stdencoding.encodetostring(b), time.hour*1000)\n            asyncreader.close()\n            time.sleep(1 * time.second)\n            msg <- fmt.sprint(i, \" exit running ... \", time.now())\n            wg.done()\n        }(index)\n\n        msg <- fmt.sprint(index, \" \", index%tenthousand == 0, \" after go call\")\n        if index%tenthousand == 0 {\n            wg.wait()\n            msg <- fmt.sprint(\"..\", index, \" done waiting after go call. \", time.now())\n        }\n    }\n    msg <- \"bye forever.\"\n\n    wg.wait()\n    close(msg)\n    wgmsg.wait()\n}\n\n\/\/ just for the go playground:\nconst tenthousand = 2\n\ntype myclient struct {\n}\n\nfunc (p *myclient) set(ctx context.context, a, b string, t time.duration) {\n    \/\/ fmt.println(\"a =\", a, \", b =\", b, \", t =\", t)\n    if ctx.err() != nil {\n        fmt.println(ctx.err())\n    }\n}\n\nvar file, myw = io.pipe()\n\nfunc init() {\n    go func() {\n        for i := 1; i <= tenthousand+1; i++ {\n            fmt.fprintf(myw, \"%d text to compress aaaaaaaaaaaaaa\\n\", i)\n        }\n        myw.close()\n    }()\n    wgmsg.add(1)\n    go func() {\n        defer wgmsg.done()\n        for s := range msg {\n            fmt.println(s)\n        }\n    }()\n}\n\nvar msg = make(chan string, 100)\nvar wgmsg sync.waitgroup<\/pre>\n
\u8f93\u51fa\uff1a<\/p>\n
1 false after go call\n2 true after go call\n1 Enter running ... 2009-11-10 23:00:00 +0000 UTC m=+0.000000001\n2 Enter running ... 2009-11-10 23:00:00 +0000 UTC m=+0.000000001\n1 Exit running ... 2009-11-10 23:00:01 +0000 UTC m=+1.000000001\n2 Exit running ... 2009-11-10 23:00:01 +0000 UTC m=+1.000000001\n..2 Done waiting after go call. 2009-11-10 23:00:01 +0000 UTC m=+1.000000001\n3 false after go call\n3 Enter running ... 2009-11-10 23:00:01 +0000 UTC m=+1.000000001\n4 Done not waiting with err: EOF 2009-11-10 23:00:01 +0000 UTC m=+1.000000001\n3 Exit running ... 2009-11-10 23:00:02 +0000 UTC m=+2.000000001\nBye forever.<\/pre>\n
\u597d\u4e86\uff0c\u672c\u6587\u5230\u6b64\u7ed3\u675f\uff0c\u5e26\u5927\u5bb6\u4e86\u89e3\u4e86\u300aio.Pipe() \u5bfc\u81f4 WaitGroup \u5361\u4f4f\u300b\uff0c\u5e0c\u671b\u672c\u6587\u5bf9\u4f60\u6709\u6240\u5e2e\u52a9\uff01\u5173\u6ce8\u516c\u4f17\u53f7\uff0c\u7ed9\u5927\u5bb6\u5206\u4eab\u66f4\u591aGolang\u77e5\u8bc6\uff01<\/p>\n","protected":false},"excerpt":{"rendered":"
\u5f53\u524d\u4f4d\u7f6e\uff1a > > ...<\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"closed","ping_status":"","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[4925],"tags":[],"class_list":["post-207039","post","type-post","status-publish","format-standard","hentry","category-4925"],"_links":{"self":[{"href":"https:\/\/server.hk\/cnblog\/wp-json\/wp\/v2\/posts\/207039","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/server.hk\/cnblog\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/server.hk\/cnblog\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/server.hk\/cnblog\/wp-json\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/server.hk\/cnblog\/wp-json\/wp\/v2\/comments?post=207039"}],"version-history":[{"count":0,"href":"https:\/\/server.hk\/cnblog\/wp-json\/wp\/v2\/posts\/207039\/revisions"}],"wp:attachment":[{"href":"https:\/\/server.hk\/cnblog\/wp-json\/wp\/v2\/media?parent=207039"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/server.hk\/cnblog\/wp-json\/wp\/v2\/categories?post=207039"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/server.hk\/cnblog\/wp-json\/wp\/v2\/tags?post=207039"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}