基于 Go 的 grpc 服务器流不断堆积对 go 客户端的响应

2024-04-05 00:04:57 562人浏览泡泡鱼

摘要

小伙伴们对golang编程感兴趣吗？是否正在学习相关知识点？如果是，那么本文《基于 Go 的 grpc 服务器流不断堆积对 go 客户端的响应》，就很适合你，本篇文章讲解的知识点主要包括。在之后的文

小伙伴们对golang编程感兴趣吗？是否正在学习相关知识点？如果是，那么本文《基于 Go 的 grpc 服务器流不断堆积对 go 客户端的响应》，就很适合你，本篇文章讲解的知识点主要包括。在之后的文章中也会多多分享相关知识点，希望对大家的知识积累有所帮助！

问题内容

抱歉，如果这是一个菜鸟问题，我是 grpc 的服务器端流媒体新手。

我现在在服务器上流向客户端的函数中拥有什么

req, err := Http.newrequest("get", actualurl, nil)

//跳过一些行// res, _ := http.defaultclient.do(req)

// closing body
defer res.body.close()

body, err := ioutil.readall(res.body)

// 跳过一些行//

// unmarshaling the xml data received from the get request done above
xmldataerr := xml.unmarshal(body, &arrdata)

// creating a variable of a custom type and creating object of specific struct flight
var flightarrdata arrflights
currarrdata := flightarrdata.arr.body.flight

log.println("计算要发送到客户端的总行数", len(currarrdata))

// 循环响应并将其流发送到客户端

for i := 范围 currarrdata {

farr := &pb.flightarrivalsresponse{
        arrivalresponse: &pb.arrivalflightdata{
            hapt:     currarrdata[i].hapt,
            fltnr:    currarrdata[i].fltnr,
            sdt:      currarrdata[i].sdt,
            acreg:    currarrdata[i].acreg,
            park:     currarrdata[i].park,
            estd:     currarrdata[i].estd,
            gate:     currarrdata[i].gate,
            ablkd:    currarrdata[i].ablkd,
            actd:     currarrdata[i].actd,
            callsign: currarrdata[i].callsign,
        },
    }

    senderr := stream.send(farr)

    // skipping some lines //
}

// 完成后返回 nil 返回零 }

我面临的问题 在客户端，我有接收此响应的函数，休眠 n 分钟并再次请求响应。

客户端确实在第一次调用中得到了预期的响应，但是对于每个后续调用都会发生一些奇怪的事情，这就是我当前的问题，我尝试以下面每个后续调用的形式进行说明：

从客户端到服务器的调用 1 --> 服务器返回 200 行

客户端睡眠 n 分钟

从客户端到服务器的调用 2 --> 服务器返回 400 行！（基本上每行都是 200 + 200 的两倍）

客户端睡眠 n 分钟

从客户端到服务器的调用 3 --> 服务器返回 600 行！！ (200+200+200)

摘要

我确实在客户端检查 error == io.eof，现在阻止从服务器到客户端的响应堆积的唯一方法是停止服务器并重新启动。

我不确定我在这里错过了什么，以确保我只发送从 get 请求收到的实际且准确的响应。非常感谢对此的任何提示。

更多信息

来自 proto 文件中 grpc protobuffer def 的部分

rpc getflightarrivals (flightarrivalsrequesturl) returns (stream flightarrivalsresponse) {
    };

上述grpc服务器端实现的完整代码

func (s *server) getflightarrivals(url *pb.flightarrivalsrequesturl, stream pb.flight_getflightarrivalsserver) error {

    cdata := ch.configprocessor() // fetches the initial part of the url from config file

    if url.arrivalurl == "" {
        actualurl = cdata.furl + "/arr/all" // adding the arrival endpoint

    } else {
        actualurl = url.arrivalurl
    }

    // build new request to get arrival data
    req, err := http.newrequest("get", actualurl, nil)
    if err != nil {
        log.fatalln("recheck url or connectivity, failed to make rest call")
    }
    req.header.add("cache-control", "no-cache")
    req.header.add("accept", "text/plain")
    req.header.add("connection", "keep-alive")
    req.header.add("app_id", cdata.appid)
    req.header.add("app_key", cdata.appkey)
    req.header.add("content-type", "application/xml")


    res, _ := http.defaultclient.do(req)

    // closing body
    defer res.body.close()

    body, err := ioutil.readall(res.body)
    if err != nil {
        log.fatalln("failed to get any response")
        return err
    }
    // unmarshaling the xml data
    xmldataerr := xml.unmarshal(body, &flightarrdata)
    if xmldataerr != nil {
        log.fatalln("failed to unmarshal arrival xml data see error, ", xmldataerr)
        return xmldataerr
    }

    currarrdata := flightarrdata.arr.body.flight
    log.println("counting total arrivals in finland", len(currarrdata))
    log.println("starting flightdeparturesresponse for client")
    for i := range currarrdata {

        farr := &pb.flightarrivalsresponse{
            arrivalresponse: &pb.arrivalflightdata{
                hapt:     currarrdata[i].hapt,
                fltnr:    currarrdata[i].fltnr,
                sdt:      currarrdata[i].sdt,
                acreg:    currarrdata[i].acreg,
                park:     currarrdata[i].park,
                estd:     currarrdata[i].estd,
                gate:     currarrdata[i].gate,
                ablkd:    currarrdata[i].ablkd,
                actd:     currarrdata[i].actd,
                callsign: currarrdata[i].callsign,
            },
        }

        senderr := stream.send(farr)

        if senderr != nil {
            log.fatalln("failed to stream arrival response to the client, see error ", senderr)
            return senderr
        }

    }
    currarrdata = nil
    log.println("attempting to empty the arrival data")
    return nil
}

用于检查 grpc 的上述实现的测试用例

我刚刚启动一个测试 grpc 服务器并在这个测试用例中调用该 grpc。客户端的实现与接收数据相同。

func TestGetFlightArrivals(t *testing.T) {
    const addr = "localhost:50051"
    conn, err := grpc.Dial(addr, grpc.WithInsecure())
    if err != nil {
        t.Fatalf("Did not connect: #{err}")
    }
    defer conn.Close()
    f := pb.NewFlightClient(conn)

    t.Run("GetFlightArrivals", func(t *testing.T) {

        var res, err = f.GetFlightArrivals(context.Background(), &pb.FlightArrivalsRequestURL{ArrivalURL: ""})
        if err != nil {
            t.Error("Failed to make the REST call", err)
        }

        for {
            msg, err := res.Recv()
            if err == io.EOF {
                t.Log("Finished reading all the message")
                break
            }
            if err != nil {
                t.Error("Failed to receive response")
            }
            t.Log("Message from the server", msg.GetArrivalResponse())

        }
    })
}