Асинхронная работа после ответа

Question

Я пытаюсь реализовать http-сервер, который:

• Рассчитать дальнейшее перенаправление, используя некоторую логику
• Перенаправить пользователя
• Журнал пользовательских данных

Целью является достижение максимальной пропускной способности (не менее 15 к / с). Для этого я хочу асинхронно сохранять журнал. Я использую Кафку в качестве системы регистрации и отдельный блок регистрации кода в отдельной программе. Общий пример текущей реализации:

package main

import (
    "github.com/confluentinc/confluent-kafka-go/kafka"
    "net/http"
    "time"
    "encoding/json"
)

type log struct {
    RuntimeParam  string `json:"runtime_param"`
    AsyncParam    string `json:"async_param"`
    RemoteAddress string `json:"remote_address"`
}

var (
    producer, _ = kafka.NewProducer(&kafka.ConfigMap{
        "bootstrap.servers": "localhost:9092,localhost:9093",
        "queue.buffering.max.ms": 1 * 1000,
        "go.delivery.reports": false,
        "client.id": 1,
    })
    topicName = "log"
)

func main() {
    siteMux := http.NewServeMux()
    siteMux.HandleFunc("/", httpHandler)
    srv := &http.Server{
        Addr: ":8080",
        Handler: siteMux,
        ReadTimeout:  2 * time.Second,
        WriteTimeout: 5 * time.Second,
        IdleTimeout:  10 * time.Second,
    }
    if err := srv.ListenAndServe(); err != nil {
        panic(err)
    }
}

func httpHandler(w http.ResponseWriter, r *http.Request) {
    handlerLog := new(log)
    handlerLog.RuntimeParam = "runtimeDataString"
    http.Redirect(w, r, "http://google.com", 301)
    go func(goroutineLog *log, request *http.Request) {
        goroutineLog.AsyncParam = "asyncDataString"
        goroutineLog.RemoteAddress = r.RemoteAddr
        jsonLog, err := json.Marshal(goroutineLog)
        if err == nil {
            producer.ProduceChannel() <- &kafka.Message{
                TopicPartition: kafka.TopicPartition{Topic: &topicName, Partition: kafka.PartitionAny},
                Value:          jsonLog,
            }
        }
    }(handlerLog, r)
}

Вопросы:

1. Правильно ли / эффективно использовать отдельную процедуру для реализации асинхронного ведения журнала или я должен использовать другой подход? (рабочие и каналы, например)
2. Может быть, есть способ еще улучшить производительность сервера, который мне не хватает?

Answer 1

1. Да, это правильное и эффективное использование горутина (как указано в комментариях Хлипкий ). Я не мог согласиться с этим, это хороший подход.

Проблема в том, что обработчик может завершить выполнение до того, как программа запустит обработку всего, и запрос (который является указателем) может быть пропущен или у вас могут быть некоторые скачки вниз по стеку промежуточного программного обеспечения. Я читаю ваши комментарии, что это не ваш случай, но в целом вы не должны передавать запрос в программу. Как видно из вашего кода, вы на самом деле используете только RemoteAddr из запроса, и почему бы не перенаправить сразу и не включить запись в оператор defer? Итак, я бы немного переписал ваш обработчик:

func httpHandler(w http.ResponseWriter, r *http.Request) {
    http.Redirect(w, r, "http://google.com", 301)
    defer func(runtimeDataString, RemoteAddr string) {
            handlerLog := new(log)
            handlerLog.RuntimeParam = runtimeDataString
            handlerLog.AsyncParam = "asyncDataString"
            handlerLog.RemoteAddress = RemoteAddr
            jsonLog, err := json.Marshal(handlerLog)
            if err == nil {
                producer.ProduceChannel() <- &kafka.Message{
                    TopicPartition: kafka.TopicPartition{Topic: &topicName, Partition: kafka.PartitionAny},
                    Value:          jsonLog,
                }
            }
        }("runtimeDataString", r.RemoteAddr)
}  

2. Мгновенные процедуры вряд ли улучшат производительность вашего сервера, так как вы просто отправляете ответ раньше, и эти соединения kafka могут накапливаться в фоновом режиме и замедлять работу всего сервера. Если вы считаете это узким местом, вы можете рассмотреть возможность сохранения журналов локально и отправки их в kafka в другом процессе (или пуле работников) за пределами вашего сервера. Это может распределить рабочую нагрузку с течением времени (например, отправлять меньше журналов, когда у вас больше запросов, и наоборот).