GAE Golang实时通知去重：利用Memcache原子操作确保唯一处理

package main

import (
    "context"
    "crypto/sha256"
    "encoding/hex"
    "fmt"
    "log"
    "net/http"
    "time"

    "google.golang.org/appengine"
    "google.golang.org/appengine/memcache"
)

// NotificationPayload 模拟第三方通知的数据结构
type NotificationPayload struct {
    ID      string `json:"id"`
    Content string `json:"content"`
    // ... 其他通知字段
}

// generateUniqueKey 根据通知内容生成一个唯一的Memcache键
// 实际应用中，如果第三方提供唯一ID，应优先使用
func generateUniqueKey(payload NotificationPayload) string {
    // 示例：使用ID和内容生成SHA256哈希作为键
    // 更健壮的方式是结合通知类型、源等信息
    data := []byte(payload.ID + ":" + payload.Content)
    hash := sha256.Sum256(data)
    return "notification_dedupe:" + hex.EncodeToString(hash[:])
}

func handleNotification(w http.ResponseWriter, r *http.Request) {
    ctx := appengine.NewContext(r)

    // 1. 解析请求体，获取通知数据
    // 实际应用中需要进行JSON解析等
    // 简化处理，假设我们已经从请求中获取到NotificationPayload
    // 例如：
    // var payload NotificationPayload
    // if err := json.NewDecoder(r.Body).Decode(&payload); err != nil {
    //     http.Error(w, "Invalid request body", http.StatusBadRequest)
    //     return
    // }
    // 为示例创建一个模拟的payload
    mockPayload := NotificationPayload{
        ID:      "unique_notification_id_123",
        Content: "This is a real-time notification message.",
    }

    // 2. 生成唯一的Memcache键
    dedupeKey := generateUniqueKey(mockPayload)

    // 3. 尝试使用memcache.Add进行去重
    item := &memcache.Item{
        Key:        dedupeKey,
        Value:      []byte("processing"), // 值不重要，只要存在即可
        Expiration: time.Second * 10,     // 设置一个短期的过期时间，防止键永久存在
    }

    err := memcache.Add(ctx, item)
    if err == nil {
        // 成功添加到Memcache，说明是首次处理此通知
        log.Printf(ctx, "Processing unique notification: %s", mockPayload.ID)

        // 在这里执行实际的业务逻辑，例如：
        // - 写入数据存储 (Datastore)
        // - 发布到消息队列 (Pub/Sub)
        // - 调用其他服务
        // ...
        time.Sleep(time.Millisecond * 50) // 模拟处理耗时

        log.Printf(ctx, "Notification processed successfully: %s", mockPayload.ID)
        w.WriteHeader(http.StatusOK)
        fmt.Fprintf(w, "Notification %s processed.", mockPayload.ID)

        // 可选：如果确定处理完成且不再需要去重，可以删除Memcache键
        // memcache.Delete(ctx, dedupeKey)

    } else if err == memcache.ErrNotStored {
        // 键已存在，说明是重复通知，直接忽略
        log.Printf(ctx, "Ignoring duplicate notification: %s (Key: %s)", mockPayload.ID, dedupeKey)
        w.WriteHeader(http.StatusAccepted) // 或者 StatusOK，表示请求已接收但未处理
        fmt.Fprintf(w, "Notification %s is a duplicate and ignored.", mockPayload.ID)
    } else {
        // Memcache操作发生其他错误
        log.Errorf(ctx, "Memcache error for key %s: %v", dedupeKey, err)
        http.Error(w, "Internal server error during deduplication", http.StatusInternalServerError)
    }
}

func main() {
    http.HandleFunc("/notify", handleNotification)
    appengine.Main() // 启动GAE应用
}