Google Pub/Sub订阅者重试策略及Go代码逻辑咨询

我是Google Pub/Sub新手，用Go开发，作为第三方订阅者（无订阅权限，仅负责消费），合作方每小时推送100条消息，需要持续消费。现有一段后台拉取消息的代码：

func PullMyMessages() {
   ctx := context.Background()
   client, _ := pubsub.NewClient(ctx, "myAPI", option.WithCredentialsFile("abc.json"))
   sub := client.Subscription("Mysub")
   msgSlice := make(chan *pubsub.Message, 1)
   cctx, cancel := context.WithCancel(context.TODO())
   go sub.Receive(cctx, func(ctx context.Context, m *pubsub.Message) {
      msgSlice <- m
   })
   for {
      select {
      case res := <-msgSlice:
         fmt.Printf("Got message: %q\n", string(res.Data))
         res.Ack()

      case <-time.After(5 * time.Minute):
         cancel()
      }
   }
}

目前代码可正常拉取消息，存在两个问题：

1. 作为仅订阅者，能否为处理失败的消息设置重试策略？求示例代码或相关指引。
2. 当前代码是否会在后台运行，且5分钟未收到消息时自动退出？

实际场景中将移除时间限制以持续运行，但需实现消息重试机制。

问题1：处理失败消息的重试策略设置 #

作为仅拥有消费权限的订阅者，无法直接修改订阅层面的重试配置（比如重试间隔、最大重试次数、死信队列这类需要订阅的创建/修改权限），但可以通过消费端代码实现两种重试逻辑：

方案1：依赖Pub/Sub原生重试 #

当消息处理失败时，不要调用Ack()，而是调用Nack()，Pub/Sub会自动将消息重新放回订阅队列，按照订阅所有者预先配置的重试策略进行重推。

方案2：本地自定义重试+Pub/Sub兜底 #

如果需要在本地先尝试几次后再交给服务端重推，可以在消息处理逻辑里添加本地重试循环，示例代码如下：

import (
	"context"
	"fmt"
	"log"
	"time"

	"cloud.google.com/go/pubsub"
	"google.golang.org/api/option"
)

func PullMyMessages() {
	ctx := context.Background()
	client, err := pubsub.NewClient(ctx, "myAPI", option.WithCredentialsFile("abc.json"))
	if err != nil {
		log.Fatalf("Failed to create pubsub client: %v", err)
	}
	defer client.Close()

	sub := client.Subscription("Mysub")
	// 调整接收配置，控制并发处理数
	sub.ReceiveSettings.MaxOutstandingMessages = 10

	cctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// 直接在Receive回调中处理消息，无需额外channel
	err = sub.Receive(cctx, func(ctx context.Context, m *pubsub.Message) {
		maxLocalRetries := 3
		var processErr error

		// 本地指数退避重试
		for retryCount := 0; retryCount < maxLocalRetries; retryCount++ {
			processErr = processMessage(m.Data)
			if processErr == nil {
				m.Ack()
				return
			}
			log.Printf("Message retry %d/%d failed: %v", retryCount+1, maxLocalRetries, processErr)
			// 指数退避：第n次重试等待2^n秒
			time.Sleep(time.Duration(1<<retryCount) * time.Second)
		}

		// 本地重试全部失败，交给Pub/Sub重推
		log.Printf("All local retries exhausted, nacking message: %v", processErr)
		m.Nack()
	})

	if err != nil {
		log.Fatalf("Pub/Sub receive error: %v", err)
	}
}

// 模拟业务消息处理函数
func processMessage(data []byte) error {
	content := string(data)
	// 模拟处理失败场景
	if content == "error_case" {
		return fmt.Errorf("invalid content: %s", content)
	}
	fmt.Printf("Successfully processed message: %q\n", content)
	return nil
}

说明：

• 本地重试建议用指数退避，避免短时间内重复请求占用资源；
• 调用Nack()后，消息会按照订阅本身的配置重新推送，直到达到订阅设置的重试上限（若配置了死信队列，最终会转入死信队列）。

问题2：代码的后台运行与退出逻辑 #

当前代码确实会在后台运行，并且连续5分钟未收到消息时会自动退出：

• sub.Receive()在goroutine中启动，主goroutine进入for循环的select分支；
• 当5分钟内没有消息从msgSlice传来时，time.After(5 * time.Minute)会触发，调用cancel()会终止sub.Receive()的上下文，导致接收停止，最终函数退出。

如果要实现持续运行，只需移除time.After分支即可：

for {
   select {
   case res := <-msgSlice:
      fmt.Printf("Got message: %q\n", string(res.Data))
      // 这里加入你的处理和重试逻辑
      res.Ack()
   }
}

更简洁的方式是直接在sub.Receive()的回调中处理消息（如问题1的示例），避免额外channel带来的goroutine调度开销。

内容的提问来源于stack exchange，提问作者Chris