我正在使用 wrk 对一个用 Go 编写的简单网络服务器进行基准测试。服务器在具有 4GB RAM 的机器上运行。在测试开始时，代码服务高达 2000 个请求/秒，性能非常好。但随着时间的推移，进程使用的内存逐渐增加，一旦达到 85%(我正在使用 top 进行检查)，吞吐量就会下降到约 100 个请求/秒。一旦我重新启动服务器，吞吐量再次增加到最佳数量。

性能下降是内存问题吗？为什么 Go 不释放这段内存？我的 Go 服务器看起来像这样:

func main() {
    defer func() {
        // Wait for all messages to drain out before closing the producer
        p.Flush(1000)
        p.Close()
    }()

    http.HandleFunc("/endpoint", handler)
    log.Fatal(http.ListenAndServe(":8080", nil))
}

在处理程序中，我将传入的 Protobuf 消息转换为 Json，并使用融合的 Kafka Go 库将其写入 Kafka。

var p, err = kafka.NewProducer(&kafka.ConfigMap{
    "bootstrap.servers": "abc-0.com:6667,abc-1.com:6667",
    "message.timeout.ms": "30000",
    "sasl.kerberos.keytab": "/opt/certs/TEST.KEYTAB",
    "sasl.kerberos.principal": "TEST@TEST.ABC.COM",
    "sasl.kerberos.service.name": "kafka",
    "security.protocol": "SASL_PLAINTEXT",
})

var topic = "test"

func handler(w http.ResponseWriter, r *http.Request) {
    body, _ := ioutil.ReadAll(r.Body)

    // Deserialize byte[] to Protobuf message
    protoMessage := &tutorial.REALTIMEGPS{}
    _ := proto.Unmarshal(body, protoMessage)

    // Convert Protobuf to Json
    realTimeJson, _ := convertProtoToJson(protoMessage)

    _, err := fmt.Fprintf(w, "")

    if err != nil {
        log.Fatal(responseErr)
    }

    // Send to Kafka
    produceMessage([]byte(realTimeJson))
}

func produceMessage(message []byte) {
    // Delivery report
    go func() {
        for e := range p.Events() {
            switch ev := e.(type) {
            case *kafka.Message:
                if ev.TopicPartition.Error != nil {
                    log.Println("Delivery failed: ", ev.TopicPartition)
                } else {
                    log.Println("Delivered message to ", ev.TopicPartition)
                }
            }
        }
    }()

    // Send message
    _ := p.Produce(&kafka.Message{
        TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny},
        Value:          message,
    }, nil)
}

func convertProtoToJson(pb proto.Message) (string, error) {
    marshaler := jsonpb.Marshaler{}
    json, err := marshaler.MarshalToString(pb)
    return json, err
}

最佳答案

问题是，在您的请求的每个结束时，您调用了 produceMessage()，它向 kafka 发送消息，并启动一个 goroutine 来接收事件检查错误。

您的代码会在传入请求进入时不间断地启动 goroutine，并且它们不会结束，直到您的 kafka 客户端出现问题。这需要越来越多的内存，并且可能需要越来越多的 CPU，因为要调度更多的 goroutine。

不要这样做。单个 goroutine 足以用于交付报告目的。设置好 p 变量后启动单个 goroutine，一切顺利。

例如:

var p *kafka.Producer

func init() {
    var err error
    p, err = kafka.NewProducer(&kafka.ConfigMap{
        // ...
    }
    if err != nil {
        // Handle error
    }

    // Delivery report
    go func() {
        for e := range p.Events() {
            switch ev := e.(type) {
            case *kafka.Message:
                if ev.TopicPartition.Error != nil {
                    log.Println("Delivery failed: ", ev.TopicPartition)
                } else {
                    log.Println("Delivered message to ", ev.TopicPartition)
                }
            }
        }
    }()
}

关于随着请求数量的增加，Go 网络服务器性能急剧下降，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/57919755/