多阅读器的并发 POST 不返回响应

Question

我有一个使用echo的概念验证 http 服务器，它接受带有 JSON 正文的 POST 请求。我正在尝试使用管道和多写入器将请求主体流式传输到多个 POST 请求，但它无法正常工作。

在下面的示例中，我可以看到数据被发送到 2 个 POST 端点，我可以看到来自这些请求的日志，但我从未收到回复，似乎代码挂起等待功能完成http.Post(...)。

如果我直接调用这两个端点，它们可以正常工作并给出有效的 json 响应，所以我相信问题出在这段代码上，它是我的路由处理程序。

func ImportAggregate(c echo.Context) error {

    oneR, oneW := io.Pipe()

    twoR, twoW := io.Pipe()

    done := make(chan bool, 2)

    go func() {

        fmt.Println("Product Starting")

        response, err := http.Post("http://localhost:1323/products/import", "application/json", oneR)

        if err != nil {

            fmt.Println(err)

        } else {

            fmt.Println(response.Body)

        }

        done <- true

    }()

    go func() {

        fmt.Println("Import Starting")

        response, err := http.Post("http://localhost:1323/discounts/import", "application/json", twoR)

        if err != nil {

            fmt.Println(err)

        } else {

            fmt.Println(response.Body)

        }

        done <- true

    }()

    mw := io.MultiWriter(oneW, twoW)

    io.Copy(mw, c.Request().Body)

    <-done

    <-done

    return c.String(200, "Imported")

}

控制台中的输出是：

Product Starting

Import Starting

Answer 1

OP 代码中的问题是 http.Post 调用从不检测所提供的 io.Reader 的 EOF。发生这种情况是因为提供的半写管道永远不会关闭，因此，半读管道永远不会发出常规的 EOF 错误。作为关于关闭半读管道会产生不规则错误的 OP 评论的注释，必须了解从关闭的管道读取不是正确的行为。因此在这种情况下，应注意在复制完内容后立即关闭半写侧。生成的源代码应更改为func ImportAggregate(c echo.Context) error {&nbsp; &nbsp; oneR, oneW := io.Pipe()&nbsp; &nbsp; twoR, twoW := io.Pipe()&nbsp; &nbsp; done := make(chan bool, 2)&nbsp; &nbsp; go func() {&nbsp; &nbsp; &nbsp; &nbsp; fmt.Println("Product Starting")&nbsp; &nbsp; &nbsp; &nbsp; response, err := http.Post("http://localhost:1323/products/import", "application/json", oneR)&nbsp; &nbsp; &nbsp; &nbsp; if err != nil {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; fmt.Println(err)&nbsp; &nbsp; &nbsp; &nbsp; } else {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; fmt.Println(response.Body)&nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; &nbsp; &nbsp; done <- true&nbsp; &nbsp; }()&nbsp; &nbsp; go func() {&nbsp; &nbsp; &nbsp; &nbsp; fmt.Println("Import Starting")&nbsp; &nbsp; &nbsp; &nbsp; response, err := http.Post("http://localhost:1323/discounts/import", "application/json", twoR)&nbsp; &nbsp; &nbsp; &nbsp; if err != nil {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; fmt.Println(err)&nbsp; &nbsp; &nbsp; &nbsp; } else {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; fmt.Println(response.Body)&nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; &nbsp; &nbsp; done <- true&nbsp; &nbsp; }()&nbsp; &nbsp; mw := io.MultiWriter(oneW, twoW)&nbsp; &nbsp; io.Copy(mw, c.Request().Body)&nbsp; &nbsp; oneW.Close()&nbsp; &nbsp; twoW.Close()&nbsp; &nbsp; <-done&nbsp; &nbsp; <-done&nbsp; &nbsp; return c.String(200, "Imported")}OP 问题之外的旁注：必须围绕 io.Copy 实施错误检查以检测传输错误。不需要关闭管道的半读端，http.Post 将在收到 EOF 信号后执行此操作。必须在复制输入请求之前声明并启动负责使用管道的 goroutine。Pipes 是同步的，代码会在 io.Copy 等待另一端被消耗的过程中阻塞。done chan 不需要无缓冲（长度为 2）一种将错误从传出请求转发到传出响应的方法是使用 type 的通道(chan error)，循环两次，并检查遇到的第一个错误。