Google Cloud Platform：Cloud Function处理Pub/Sub消息时如何忽略新消息？

要实现「处理消息期间忽略新的“done”消息、不启动新实例」，核心是给你的Cloud Function加一个分布式锁，确保同一时间只有一个实例在处理业务逻辑，新消息过来时拿不到锁就直接确认（ACK）消息，相当于忽略。

下面给你两种适合GCP场景的实现方式，附Go代码示例：

方式一：用Cloud Memorystore（Redis）做锁 #

Redis的SETNX命令天生适合做分布式锁，配合过期时间可以避免锁一直占用：

import (
    "context"
    "time"
    "github.com/go-redis/redis/v8"
)

var redisClient *redis.Client

// 初始化Redis客户端，部署前替换成你的实例地址
func init() {
    redisClient = redis.NewClient(&redis.Options{
        Addr:     "your-redis-instance-ip:6379",
        Password: "", // 若Redis设了密码请填写
        DB:       0,
    })
}

func HandleDoneMessage(ctx context.Context, m pubsub.Message) error {
    lockKey := "done-function-lock"
    // 尝试获取锁，过期时间设为函数最长处理时间+缓冲（比如5分钟）
    acquired, err := redisClient.SetNX(ctx, lockKey, "locked", 300*time.Second).Result()
    if err != nil {
        return err
    }

    if !acquired {
        // 锁已被占用，直接ACK消息，不处理
        return nil
    }

    // 确保函数结束后释放锁，即使业务逻辑出错
    defer func() {
        _, _ = redisClient.Del(ctx, lockKey).Result()
    }()

    // --------------------------
    // 这里写你的"done"消息处理逻辑
    // --------------------------

    return nil
}

方式二：用Cloud Firestore做锁 #

如果不想额外部署Redis，用Firestore的文档锁也可以，借助事务和TTL自动过期：

import (
    "context"
    "fmt"
    "cloud.google.com/go/firestore"
)

var firestoreClient *firestore.Client

// 初始化Firestore客户端
func init() {
    var err error
    firestoreClient, err = firestore.NewClient(context.Background(), "your-gcp-project-id")
    if err != nil {
        panic(err) // 初始化失败直接退出
    }
}

func HandleDoneMessage(ctx context.Context, m pubsub.Message) error {
    lockDoc := firestoreClient.Collection("function-locks").Doc("done-processing-lock")

    // 用事务尝试获取锁
    err := firestoreClient.RunTransaction(ctx, func(ctx context.Context, tx *firestore.Transaction) error {
        doc, err := tx.Get(lockDoc)
        if err != nil && err != firestore.ErrNotFound {
            return err
        }
        if doc.Exists() {
            // 锁已存在，返回自定义错误
            return fmt.Errorf("lock occupied")
        }
        // 创建锁文档，设置300秒后自动过期
        return tx.Set(lockDoc, map[string]interface{}{
            "in_use": true,
        }, firestore.SetTTL(300*time.Second))
    })

    if err != nil {
        if err.Error() == "lock occupied" {
            // 忽略消息，直接ACK
            return nil
        }
        return err
    }

    // 函数结束后手动释放锁
    defer func() {
        _, _ = lockDoc.Delete(ctx)
    }()

    // --------------------------
    // 这里写你的"done"消息处理逻辑
    // --------------------------

    return nil
}

关键注意事项 #

• 过期时间一定要设得比你的函数最长处理时间长，避免锁提前过期导致多个实例同时运行。
• 必须用defer确保锁的释放，哪怕业务逻辑抛出错误，也不会让锁一直占用。
• 不要依赖Cloud Function的实例数限制（比如设置maxInstances: 1），因为Pub/Sub的消息推送机制可能在实例 busy 时仍会尝试推送，锁是最可靠的方案。

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