管理 golang 中的 goroutine 并接收响应

在 Go 语言中管理 goroutine 是并发编程的关键，本文探讨了如何通过 goroutine 获取响应并处理错误。通过使用通道和错误传播，开发者可以异步调用 goroutine，同时在主线程中接收结果或错误信息。这使得应用程序可以响应事件，处理流数据，并在 goroutine 失败时提供适当的响应，从而提高并发性并增强用户体验。

我在 getposition、getstatus 和重新路由中调用了 go 例程，目前我正在主函数中调用 go getstatus，该函数具有 go func()，它处理流 grpc 和 sse 的事件。

目前这是我的代码，我尝试过

func getpositioncontext(ctx context.context, uav pb.uavcontrollerclient, uavid *pb.uavid, projectid string) {
    log.printf("getposition start")

    stream, err := uav.getposition(ctx)

    if err != nil {
        fmt.printf("error getposition:%s", err.error())
    }

    streamid, eventname := eventssubscribe(projectid, uavid.aircraft, "get_position")

    position := make(chan models.droneposition)

    // 受信ループ開始
    go func() {
        fmt.print("start getposition loop")
        for {
            msg, err := stream.recv() // msg uavposition
            if err == io.eof {
                // read done.
                fmt.print("start getposition loop closed")
                close(position)
                return
            }
            if err != nil {
                log.fatalf("failed to receive getposition : %v", err)
                close(position)
                return
            }
            // log.printf("position point[%s](%f, %f, %f) h:%f", uavid.aircraft, msg.latitude, msg.longitude, msg.altitude, msg.heading)

            waypoint := models.waypointitem{
                latitude:  msg.latitude,
                longitude: msg.longitude,
                altitude:  msg.altitude,
                heading:   msg.heading,
            }

            droneposition := models.droneposition{
                name:          uavid.aircraft,
                itemparameter: waypoint,
            }

            // publish to eventgo
            publishnotif(droneposition, streamid, eventname)
            return
        }
    }()

    startmsg := pb.uavcontrollerpositionrequest{
        uavid:       uavid,
        instruction: true,
        interval:    2,
    }

    fmt.print("send getposition start")

    if err := stream.send(&startmsg); err != nil {
        log.fatalf("failed to send getposition: %v", err)
    }

    <-position

    stream.closesend()
    fmt.print("end of getposition")
}

这是我调用这个函数的部分

go utils.getposition(ctx, uavservice, &uavid, projectid)

我想从 go func 获取返回值，就像 grpc 服务器中一切正常一样，没有问题，我应该返回 200 成功，如果失败则返回 500。

for {
        time.sleep(time.duration(60) * time.second)
    }

调用后我有这个代码，它应该每 10 秒返回一次成功

return c.JSON(500, failed or pass)

当流部分正在工作时，如果 go 例程成功或失败，我希望有一些东西返回到 ui，以便响应不会挂起，否则其他 api 调用将无法工作。

解决方案

如果你想知道在 getpositioncontext 中处理 go func() 时是否出现错误，那么你可以执行如下操作。我修改了所有错误以冒泡，而不是调用 log.fatal。顺便说一句，推迟 stream.closesend() 可能是明智的，但我缺少上下文，而且它超出了问题的范围。

我已尽力理解这个问题，但如果我遗漏了什么，请告诉我！

代码已被修改为直接在 getpositioncontext 中冒泡错误，而不是调用 log.fatal。此外，如果 go func() 中出现错误，它将通过通道发送回 getpositioncontext，然后返回给调用者。

func getpositioncontext(ctx context.context, uav pb.uavcontrollerclient, uavid *pb.uavid, projectid string) error {
    log.printf("getposition start")

    stream, err := uav.getposition(ctx)

    if err != nil {
        return fmt.errorf("error getposition: %v", err.error())
    }

    streamid, eventname := eventssubscribe(projectid, uavid.aircraft, "get_position")

    errc := make(chan error)

    // 受信ループ開始
    go func() {
        fmt.print("start getposition loop")
        for {
            msg, err := stream.recv() // msg uavposition
            if err == io.eof {
                // read done.
                fmt.print("start getposition loop closed")
                close(errc)
                return
            }
            if err != nil {
                errc <- fmt.errorf("failed to receive getposition : %v", err)
                return
            }
            // log.printf("position point[%s](%f, %f, %f) h:%f", uavid.aircraft, msg.latitude, msg.longitude, msg.altitude, msg.heading)

            waypoint := models.waypointitem{
                latitude:  msg.latitude,
                longitude: msg.longitude,
                altitude:  msg.altitude,
                heading:   msg.heading,
            }

            droneposition := models.droneposition{
                name:          uavid.aircraft,
                itemparameter: waypoint,
            }

            // publish to eventgo
            publishnotif(droneposition, streamid, eventname)

            // did you mean to return here? the for loop will only ever execute one time.
            // if you didn't mean to return, then remove this close
            close(errc)
            return
        }
    }()

    startmsg := pb.uavcontrollerpositionrequest{
        uavid:       uavid,
        instruction: true,
        interval:    2,
    }

    fmt.print("send getposition start")

    if err := stream.send(&startmsg); err != nil {
        return fmt.errorf("failed to send getposition: %v", err)
    }

    err = <- errc

    stream.closesend()
    fmt.print("end of getposition")
    return err
}

如果你想异步调用这个函数，但仍然想知道是否有错误，那么你可以执行以下操作。

errC := make(chan error)
go func() {
  errC <- GetPositionContext(..........)
}()
// Do some other stuff here
// Until you need the result
err := <- errC


// ... and Eventually when you want to return the success or failure as JSON
if err != nil {
  return c.JSON(500, failed)
}
return c.JSON(500, pass)

到这里，我们也就讲完了《管理 golang 中的 goroutine 并接收响应》的内容了。个人认为，基础知识的学习和巩固，是为了更好的将其运用到项目中，欢迎关注golang学习网公众号，带你了解更多关于的知识点！