多阅读器的并发 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 StartingImport Starting

Answer 1

OP 代码中的问题是 http.Post 调用从不检测所提供的 io.Reader 的 EOF。

发生这种情况是因为提供的半写管道永远不会关闭，因此，半读管道永远不会发出常规的 EOF 错误。

作为关于关闭半读管道会产生不规则错误的 OP 评论的注释，必须了解从关闭的管道读取不是正确的行为。

因此在这种情况下，应注意在复制完内容后立即关闭半写侧。

生成的源代码应更改为

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)

    oneW.Close()

    twoW.Close()

    <-done

    <-done

    return c.String(200, "Imported")

}

OP 问题之外的旁注：

• 必须围绕 io.Copy 实施错误检查以检测传输错误。

• 不需要关闭管道的半读端，http.Post 将在收到 EOF 信号后执行此操作。

• 必须在复制输入请求之前声明并启动负责使用管道的 goroutine。Pipes 是同步的，代码会在 io.Copy 等待另一端被消耗的过程中阻塞。

• done chan 不需要无缓冲（长度为 2）

• 一种将错误从传出请求转发到传出响应的方法是使用 type 的通道(chan error)，循环两次，并检查遇到的第一个错误。