如何让K8s多副本Pod内Golang应用的CronJob仅执行一次？

针对你在K8s多Pod环境下Golang内部CronJob重复发邮件的问题，这里给你几个不用拆分服务、在应用内部解决的靠谱方案：

1. 基于K8s Leader Election（选主）实现单Pod执行 #

这是最贴合K8s生态的方案：在应用内部集成选主机制，只有被选为Leader的Pod才启动CronJob，其他Pod只监听Leader状态变化，不执行任务。不管副本数多少，永远只有一个Pod在跑定时任务。

实现步骤 #

• 引入k8s.io/client-go/tools/leaderelection及相关依赖
• 用ConfigMap作为锁对象初始化Leader选举客户端
• 当选Leader后启动CronJob，失去Leader身份时优雅停止任务

代码示例 #

import (
    "context"
    "os"
    "time"

    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/client-go/kubernetes"
    "k8s.io/client-go/tools/leaderelection"
    "k8s.io/client-go/tools/leaderelection/resourcelock"
    "github.com/robfig/cron/v3"
    "k8s.io/client-go/rest"
)

// 启动Leader选举并管理CronJob
func startLeaderElection(client *kubernetes.Clientset, podName, namespace string) {
    lock := &resourcelock.ConfigMapLock{
        ConfigMapMeta: metav1.ObjectMeta{
            Name:      "weekly-email-leader-lock",
            Namespace: namespace,
        },
        Client: client.CoreV1(),
        LockConfig: resourcelock.ResourceLockConfig{
            Identity: podName, // 用Pod名称作为唯一标识
        },
    }

    var cronInstance *cron.Cron

    leaderelection.RunOrDie(context.Background(), leaderelection.LeaderElectionConfig{
        Lock:          lock,
        LeaseDuration: 15 * time.Second,
        RenewDeadline: 10 * time.Second,
        RetryPeriod:   2 * time.Second,
        OnStartedLeading: func(ctx context.Context) {
            // 当选Leader后启动CronJob
            cronInstance = cron.New()
            _, err := cronInstance.AddFunc("@weekly", sendWeeklyEmail)
            if err != nil {
                panic(err)
            }
            cronInstance.Run()
        },
        OnStoppedLeading: func() {
            // 失去Leader身份时停止CronJob
            if cronInstance != nil {
                cronInstance.Stop()
            }
        },
    })
}

// 邮件发送核心逻辑
func sendWeeklyEmail() {
    // 这里写你的邮件发送代码
    println("Sending weekly email from leader pod...")
}

func main() {
    // 初始化K8s客户端（生产环境用InClusterConfig）
    config, err := rest.InClusterConfig()
    if err != nil {
        panic(err)
    }
    client, err := kubernetes.NewForConfig(config)
    if err != nil {
        panic(err)
    }

    // 从K8s注入的环境变量获取Pod名称和命名空间
    podName := os.Getenv("POD_NAME")
    namespace := os.Getenv("POD_NAMESPACE")

    startLeaderElection(client, podName, namespace)
}

注意事项 #

• 要给Pod绑定的ServiceAccount配置ConfigMap的get、update、create权限
• 确保CronJob能在失去Leader身份时优雅停止，避免残留定时任务

2. 分布式锁（基于K8s ConfigMap） #

如果不想用持续的Leader选举，也可以在每次Cron任务触发时抢分布式锁：只有抢到锁的Pod才执行邮件发送逻辑。这种方式适合单次触发的定时任务，不需要长期维持Leader状态。

实现思路 #

• 用ConfigMap存储锁的持有者和过期时间
• 基于K8s的乐观锁（ResourceVersion）确保只有一个Pod能成功更新锁
• 设置锁的过期时间，避免Pod挂掉导致锁永久占用

代码示例 #

import (
    "context"
    "os"
    "time"

    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/apimachinery/pkg/api/errors"
    corev1 "k8s.io/api/core/v1"
    "k8s.io/client-go/kubernetes"
    "k8s.io/client-go/rest"
)

const lockConfigMapName = "weekly-email-lock"
const lockExpireDuration = 30 * time.Minute // 锁过期时间要大于任务最长执行时间

// 尝试获取分布式锁
func tryAcquireLock(client *kubernetes.Clientset, namespace, podName string) (bool, error) {
    cm, err := client.CoreV1().ConfigMaps(namespace).Get(context.TODO(), lockConfigMapName, metav1.GetOptions{})
    if err != nil {
        // 锁不存在则创建
        if errors.IsNotFound(err) {
            newCm := &corev1.ConfigMap{
                ObjectMeta: metav1.ObjectMeta{
                    Name:      lockConfigMapName,
                    Namespace: namespace,
                },
                Data: map[string]string{
                    "lock-owner":  podName,
                    "expire-time": time.Now().Add(lockExpireDuration).Format(time.RFC3339),
                },
            }
            _, err := client.CoreV1().ConfigMaps(namespace).Create(context.TODO(), newCm, metav1.CreateOptions{})
            return err == nil, err
        }
        return false, err
    }

    // 检查锁是否过期
    expireTime, err := time.Parse(time.RFC3339, cm.Data["expire-time"])
    if err != nil || time.Now().After(expireTime) {
        // 锁已过期，尝试更新
        cm.Data["lock-owner"] = podName
        cm.Data["expire-time"] = time.Now().Add(lockExpireDuration).Format(time.RFC3339)
        _, err := client.CoreV1().ConfigMaps(namespace).Update(context.TODO(), cm, metav1.UpdateOptions{})
        return err == nil, err
    }

    // 锁被其他Pod持有且未过期
    return false, nil
}

// 修改后的Cron任务逻辑
func sendWeeklyEmail() {
    // 初始化K8s客户端
    config, _ := rest.InClusterConfig()
    client, _ := kubernetes.NewForConfig(config)
    
    podName := os.Getenv("POD_NAME")
    namespace := os.Getenv("POD_NAMESPACE")

    // 尝试抢锁
    acquired, err := tryAcquireLock(client, namespace, podName)
    if err != nil {
        println("Failed to acquire lock:", err.Error())
        return
    }
    if !acquired {
        println("Lock held by another pod, skipping email send")
        return
    }

    // 抢到锁，执行邮件发送
    println("Acquired lock, sending weekly email...")
    // ... 你的邮件发送代码 ...
}

注意事项 #

• 同样需要给ServiceAccount配置ConfigMap的读写权限
• 锁的过期时间要足够长，避免任务还在执行就被其他Pod抢锁

3. Redis分布式锁（备选） #

如果你的集群已经部署了Redis，也可以用Redis实现分布式锁，逻辑和上面类似：任务触发前先抢锁，抢到才执行。这种方式不需要依赖K8s客户端，适合不想引入client-go的场景。

代码示例（用go-redis） #

import (
    "context"
    "os"
    "time"

    "github.com/go-redis/redis/v8"
    "github.com/robfig/cron/v3"
)

func sendWeeklyEmail() {
    // 初始化Redis客户端
    rdb := redis.NewClient(&redis.Options{
        Addr: "redis-service:6379", // 替换成你的Redis服务地址
    })

    lockKey := "weekly-email-lock"
    lockValue := os.Getenv("POD_NAME")
    lockExpire := 30 * time.Minute

    // 尝试获取锁
    acquired, err := rdb.SetNX(context.TODO(), lockKey, lockValue, lockExpire).Result()
    if err != nil || !acquired {
        println("Failed to acquire Redis lock, skipping")
        return
    }

    // 执行邮件发送
    println("Sending weekly email with Redis lock...")
    // ... 你的邮件发送代码 ...

    // 任务完成后手动释放锁（可选，锁会自动过期）
    defer rdb.Del(context.TODO(), lockKey)
}

func main() {
    c := cron.New()
    c.AddFunc("@weekly", sendWeeklyEmail)
    c.Run()
}

方案选择建议 #

• 如果你的CronJob是长期运行的（比如每分钟执行一次），优先选Leader Election，逻辑更稳定，避免频繁抢锁的开销
• 如果是低频率任务（比如每周一次），ConfigMap分布式锁更轻量，不需要维持Leader状态
• 如果已经有Redis集群，Redis分布式锁是不错的备选，不需要依赖K8s客户端

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