IT行业相对于一般传统行业，发展更新速度更快，一旦停止了学习，很快就会被行业所淘汰。所以我们需要踏踏实实的不断学习，精进自己的技术，尤其是初学者。今天编程网给大家整理了《在GCP Cloud Pub/Sub中逐条发布消息》，聊聊，我们一起来看看吧！
问题内容我正在开发一个示例项目，该项目从 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
但他们实际上并没有在示例中调用它，所以我不知道应该如何完成。
旁注并进一步证明我的观点，它不起作用，如果我在 topic.publishsettings var 中设置 delaythreshold 说，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 中看到这一点，其中 stackdriver 中的消息/秒大约为 10-16/秒，并且 wireshark 显示每条发送的消息的新连接。
解决方案
这可能是因为当您致电时
topic.PublishSettings = pubsub.PublishSettings{
// 字节阈值: 5000,
计数阈值：1000，
// DelayThreshold: 1000 * time.Millisecond,
}
</p>
<p>您正在将发布设置重置为零初始化结构。这会将 topic.PublishSettings.ByteThreshold 设置为 0，这意味着所有消息将立即发布；你告诉它等到它有 0 字节，它总是有 0 字节。</p>
<p>相反，您应该执行以下操作来设置 CountThreshold：</p>
<p>topic.PublishSettings.CountThreshold = 1000
这同样适用于其他字段。它们已经按照 here 的描述初始化为默认值，如果您想更改它们，请直接修改它们，而不是重新分配整个 PublishSetttings 对象。
今天关于《在GCP Cloud Pub/Sub中逐条发布消息》的内容介绍就到此结束，如果有什么疑问或者建议，可以在编程网公众号下多多回复交流；文中若有不正之处，也希望回复留言以告知！
