Я работаю над примером проекта, который берет вывод из bigquery и публикует его в pubsub.Вывод строки из bigquery может быть> 100 000.Я видел, что есть варианты для пакетной публикации, и я прочитал во многих местах, что 1k сообщений на пакет идеально.Проблема, с которой я сталкиваюсь, заключается в том, что при жизни я не могу заставить его пакетировать несколько сообщений, и я думаю, что решение простое, но мне не хватает, как это сделать ..
Вотто, что у меня есть сейчас, и все, что он делает, это публикует одно сообщение за раз.
func publish(client pubsub.Client, data []byte) (string, error) {
ctx := context.Background()
topic := client.Topic("topic-name")
topic.PublishSettings = pubsub.PublishSettings{
// ByteThreshold: 5000,
CountThreshold: 1000, // no matter what I put here it still sends one per publish
// DelayThreshold: 1000 * time.Millisecond,
}
result := topic.Publish(ctx, &pubsub.Message{
Data: data,
})
id, err := result.Get(ctx)
if err != nil {
return "", err
}
return id, nil
}
И эта функция вызывается:
for _, v := range qr {
data, err := json.Marshal(v)
if err != nil {
log.Printf("Unable to marshal %s", data)
continue
}
id, err := publish(*pubsubClient, data)
if err != nil {
log.Printf("Unable to publish message: %s", data)
}
log.Printf("Published message with id: %s", id)
}
Где qr - это фрагмент структур, которыесодержат данные, возвращаемые запросом bigquery.
Теперь, это связано с тем, как я вызываю функцию publish, которая публикует каждое сообщение, а topic.PublishSettings перезаписывается при каждом вызове метода, поэтому он забываетпредыдущие сообщения?Я в растерянности.
Я видел здесь часть кода для пакетной публикации: https://github.com/GoogleCloudPlatform/golang-samples/blob/master/pubsub/topics/main.go#L217
Но они на самом деле не называют его в своем примере, поэтому я не могурасскажи как это надо сделать.
Дополнительное замечание и еще раз доказываю, что это не работает, если я задаю DelayThreshold в переменной topic.PublishSettings, скажем, 1 секунда, он просто публикует одно сообщение каждую секунду, а не всесообщения, которые должны быть в памяти.
Благодарю за помощь, спасибо.
РЕДАКТИРОВАТЬ # 1:
Так что с комментариями kingkupps,Я включил код, чтобы он был таким для целей тестирования: (названия проектов и тем изменились с реальных)
func QueryAndPublish(w http.ResponseWriter, r *http.Request) {
ctx := context.Background()
// setting up the pubsub client
pubsubClient, err := pubsub.NewClient(ctx, "fake-project-id")
if err != nil {
log.Fatalf("Unable to get pubsub client: %v", err)
}
// init topic and settings for publishing 1000 messages in batch
topic := pubsubClient.Topic("fake-topic")
topic.PublishSettings = pubsub.PublishSettings{
// ByteThreshold: 5000,
CountThreshold: 1000,
// DelayThreshold: 1000 * time.Millisecond,
}
// bq set up
bqClient, err := bigquery.NewClient(ctx, "fake-project-id")
if err != nil {
log.Fatalf("Unable to get bq client: %v", err)
}
// bq query function call
qr, err := query(*bqClient)
if err != nil {
log.Fatal(err)
}
log.Printf("Got query results, publishing now")
// marshalling messages to json format
messages := make([][]byte, len(qr))
timeToMarshal := time.Now()
for i, v := range qr {
data, err := json.Marshal(v)
if err != nil {
log.Printf("Unable to marshal %s", data)
continue
}
messages[i] = data
}
elapsedMarshal := time.Since(timeToMarshal).Nanoseconds() / 1000000
log.Printf("Took %v ms to marshal %v messages", elapsedMarshal, len(messages))
// publishing messages
timeToPublish := time.Now()
publishCount := 0
for _, v := range messages {
// ignore result, err from topic.Publish return, just publish
topic.Publish(ctx, &pubsub.Message{
Data: v,
})
publishCount++
}
elapsedPublish := time.Since(timeToPublish).Nanoseconds() / 1000000
log.Printf("Took %v ms to publish %v messages", elapsedPublish, publishCount)
fmt.Fprint(w, "Job completed")
}
Что теперь происходит, когда мое количество сообщений составляет 100 000, оно завершает вызовы публикации впримерно 600 мс, но в фоновом режиме он будет по-прежнему публиковаться один за другим в конечной точке pubsub.
Я могу видеть это как в StackDriver, так и в Wireshark, где мои сообщения в секунду в стекдрайвере составляют примерно 10-16 в секунду, а Wireshark показывает новые соединения на каждое отправленное сообщение.