如何实现消息幂等

设计幂等的消费逻辑的关键是确保每条消息只被处理一次，即使在网络故障或消费者重启的情况下。通常使用唯一的消息ID和持久化存储来记录处理过的消息ID。

实现步骤

1. 连接kafka和redis
2. 检查消息ID
3. 处理消息
4. 标记消息已处理

package main

import (
	"context"
	"crypto/md5"
	"encoding/hex"
	"fmt"
	"github.com/confluentinc/confluent-kafka-go/kafka"
	"github.com/go-redis/redis/v8"
	"log"
	"time"
)

// 初始化Redis客户端
var ctx = context.Background()
var rdb = redis.NewClient(&redis.Options{
	Addr:     "localhost:6379",
	Password: "", // no password set
	DB:       0,  // use default DB
})

// 计算消息的唯一ID（可以使用消息的内容或其他标识）
func calculateMessageID(message []byte) string {
	hash := md5.Sum(message)
	return hex.EncodeToString(hash[:])
}

// 检查消息ID是否已处理
func isMessageProcessed(messageID string) bool {
	result, err := rdb.Get(ctx, messageID).Result()
	if err == redis.Nil {
		return false
	} else if err != nil {
		log.Fatalf("Failed to get message ID from Redis: %v", err)
	}
	return result == "processed"
}

// 标记消息ID为已处理
func markMessageAsProcessed(messageID string) {
	err := rdb.Set(ctx, messageID, "processed", 0).Err()
	if err != nil {
		log.Fatalf("Failed to set message ID in Redis: %v", err)
	}
}

// 处理消息的逻辑
func processMessage(message []byte) {
	// 在这里添加具体的消息处理逻辑
	fmt.Printf("Processing message: %s\n", string(message))
}
// 初始化Kafka消费者，读取消息，检查消息ID，处理未处理的消息，并将消息ID标记为已处理。
func main() {
	consumer, err := kafka.NewConsumer(&kafka.ConfigMap{
		"bootstrap.servers": "localhost",
		"group.id":          "myGroup",
		"auto.offset.reset": "earliest",
	})
	if err != nil {
		log.Fatalf("Failed to create consumer: %v", err)
	}
	defer consumer.Close()

	consumer.Subscribe("myTopic", nil)

	for {
		msg, err := consumer.ReadMessage(-1)
		if err == nil {
			messageID := calculateMessageID(msg.Value)
			if !isMessageProcessed(messageID) {
				processMessage(msg.Value)
				markMessageAsProcessed(messageID)
			} else {
				fmt.Printf("Message %s already processed\n", messageID)
			}
		} else {
			fmt.Printf("Consumer error: %v (%v)\n", err, msg)
		}
		time.Sleep(1 * time.Second) // 可选：添加延迟以防止消息消费过快
	}
}