我正在运行一个分析管道。

• 吞吐量为每秒约11条消息。
• 我的Pub/Sub主题包含2M预定的消息。
• 80个GCE实例正在并行拉取消息。

这是我的主题和订阅：

gcloud pubsub topics create pipeline-input

gcloud beta pubsub subscriptions create pipeline-input-sub \
    --topic pipeline-input \
    --ack-deadline 600 \
    --expiration-period never \
    --dead-letter-topic dead-letter

以下是我提取消息的方法：

import { PubSub, Message } from '@google-cloud/pubsub'

const pubSubClient = new PubSub()

const queue: Message[] = []

const populateQueue = async () => {
  const subscription = pubSubClient.subscription('pipeline-input-sub', {
    flowControl: {
      maxMessages: 5
    }
  })
  const messageHandler = async (message: Message) => {
    queue.push(message)
  }
  subscription.on('message', messageHandler)
}

const processQueueMessage = () => {
  const message = queue.shift()
  try {
    ...
    message.ack()
  } catch {
    ...
    message.nack()
  }
  processQueueMessage()
}

processQueueMessage()

处理时间约为7秒。

这是许多类似的重复案例之一。同一消息会向不同的 GCE 实例传递 5 （!!!） 次：

• 03:37:42.377
• 03:45:20.883
• 03:48:14.262
• 04:01:33.848
• 05:57:45.141

所有 5 次消息都已成功处理和 .ack（） ed。输出包含的消息比输入多 50%！我很清楚“至少一次”的行为，但我认为它可能会重复 0.01% 的消息，而不是其中的 50%。

主题输入100%没有重复。我通过云监视器验证了主题输入法和未确认消息的数量。数字匹配:在发布/订阅主题中没有重复。

更新：

< li >看起来所有这些副本都是由于ack截止日期过期而创建的。我100%确定我在600秒的期限前确认了99.9%的邮件。