// 分区选择逻辑（伪代码）
func selectPartition(record ProducerRecord) int {
    if record.Partition >= 0 {
        return record.Partition      // 1. 明确指定分区
    }
    if record.Key != nil {
        return hash(record.Key) % numPartitions  // 2. 按 Key 哈希（同Key→同分区）
    }
    return stickyPartition()         // 3. 粘性分区（默认，批次填满再切换）
}

// Go 配置示例（sarama）
config := sarama.NewConfig()
config.Producer.RequiredAcks = sarama.WaitForAll  // acks=-1
config.Producer.Retry.Max = 5
config.Producer.Return.Successes = true

config.Producer.Idempotent = true               // 开启幂等
config.Producer.RequiredAcks = sarama.WaitForAll // 幂等必须配合 acks=-1
config.Net.MaxOpenRequests = 1                   // 幂等必须配合 max.in.flight=1（单分区）

config.Producer.Transaction.ID = "my-transactional-id" // 开启事务

producer, _ := sarama.NewSyncProducer(brokers, config)

producer.BeginTxn()

producer.SendMessages([]*sarama.ProducerMessage{
    {Topic: "orders", Value: sarama.StringEncoder(`{"id":1}`)},
    {Topic: "inventory", Value: sarama.StringEncoder(`{"item":1,"count":-1}`)},
})

if err != nil {
    producer.AbortTxn()
} else {
    producer.CommitTxn()
}

// 增大心跳超时，避免因 GC/网络抖动误判 Consumer 下线
config.Consumer.Group.Session.Timeout = 30 * time.Second  // 默认10s
config.Consumer.Group.Heartbeat.Interval = 3 * time.Second // 建议 = Timeout/3

// 增大 poll 最大间隔，避免消费慢被踢出
config.Consumer.MaxProcessingTime = 5 * time.Minute // 对应 max.poll.interval.ms

# 查看消费组 Lag
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
    --group my-group --describe

# 输出示例
GROUP       TOPIC       PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
my-group    orders      0          1000            1050            50
my-group    orders      1          980             1020            40

# server.properties

# 基于时间（默认 7 天）
log.retention.hours=168

# 基于大小（超过则删除最旧的 segment）
log.retention.bytes=1073741824   # 1GB per partition

# segment 文件大小（1GB 时切分）
log.segment.bytes=1073741824

# 清理策略：delete（删除）或 compact（压缩，保留每个 Key 的最新值）
log.cleanup.policy=delete

# 推荐使用 IBM/sarama（原 Shopify/sarama）
go get github.com/IBM/sarama

# 或者使用 confluent-kafka-go（基于 librdkafka，性能更高）
go get github.com/confluentinc/confluent-kafka-go/kafka

// pkg/kafka/config.go
package kafka

import (
    "time"

    "github.com/IBM/sarama"
)

type Config struct {
    Brokers  []string
    Version  string
    ClientID string
}

func NewProducerConfig(cfg Config) *sarama.Config {
    c := sarama.NewConfig()

    // 版本
    version, _ := sarama.ParseKafkaVersion(cfg.Version) // e.g. "3.6.0"
    c.Version = version
    c.ClientID = cfg.ClientID

    // 可靠性：等待所有 ISR 确认
    c.Producer.RequiredAcks = sarama.WaitForAll
    c.Producer.Retry.Max = 5
    c.Producer.Retry.Backoff = 100 * time.Millisecond

    // 幂等生产者（防重复）
    c.Producer.Idempotent = true
    c.Net.MaxOpenRequests = 1

    // 批量压缩（提升吞吐）
    c.Producer.Compression = sarama.CompressionSnappy
    c.Producer.Flush.Bytes = 1024 * 1024  // 1MB 触发发送
    c.Producer.Flush.Frequency = 500 * time.Millisecond // 或 500ms 触发

    c.Producer.Return.Successes = true
    c.Producer.Return.Errors = true

    return c
}

func NewConsumerConfig(cfg Config) *sarama.Config {
    c := sarama.NewConfig()

    version, _ := sarama.ParseKafkaVersion(cfg.Version)
    c.Version = version
    c.ClientID = cfg.ClientID

    // 从最早的 offset 开始消费（新 group 第一次消费）
    c.Consumer.Offsets.Initial = sarama.OffsetNewest // 或 sarama.OffsetOldest

    // 手动提交 offset
    c.Consumer.Offsets.AutoCommit.Enable = false

    // Rebalance 策略：粘性，减少分区迁移
    c.Consumer.Group.Rebalance.GroupStrategies = []sarama.BalanceStrategy{
        sarama.NewBalanceStrategySticky(),
    }

    // 心跳与会话超时
    c.Consumer.Group.Session.Timeout = 30 * time.Second
    c.Consumer.Group.Heartbeat.Interval = 3 * time.Second

    return c
}

// pkg/kafka/producer.go
package kafka

import (
    "fmt"
    "time"

    "github.com/IBM/sarama"
)

type SyncProducer struct {
    producer sarama.SyncProducer
}

func NewSyncProducer(brokers []string, cfg *sarama.Config) (*SyncProducer, error) {
    p, err := sarama.NewSyncProducer(brokers, cfg)
    if err != nil {
        return nil, fmt.Errorf("new sync producer: %w", err)
    }
    return &SyncProducer{producer: p}, nil
}

// Send 发送消息，返回 partition 和 offset
func (p *SyncProducer) Send(topic, key string, value []byte) (int32, int64, error) {
    msg := &sarama.ProducerMessage{
        Topic:     topic,
        Key:       sarama.StringEncoder(key),
        Value:     sarama.ByteEncoder(value),
        Timestamp: time.Now(),
    }
    partition, offset, err := p.producer.SendMessage(msg)
    if err != nil {
        return 0, 0, fmt.Errorf("send message: %w", err)
    }
    return partition, offset, nil
}

// SendBatch 批量发送
func (p *SyncProducer) SendBatch(msgs []*sarama.ProducerMessage) error {
    if err := p.producer.SendMessages(msgs); err != nil {
        if errs, ok := err.(sarama.ProducerErrors); ok {
            for _, e := range errs {
                fmt.Printf("failed to send: %v, err: %v\n", e.Msg, e.Err)
            }
        }
        return err
    }
    return nil
}

func (p *SyncProducer) Close() error {
    return p.producer.Close()
}

// pkg/kafka/async_producer.go
package kafka

import (
    "log"
    "sync"

    "github.com/IBM/sarama"
)

type AsyncProducer struct {
    producer sarama.AsyncProducer
    wg       sync.WaitGroup
}

func NewAsyncProducer(brokers []string, cfg *sarama.Config) (*AsyncProducer, error) {
    p, err := sarama.NewAsyncProducer(brokers, cfg)
    if err != nil {
        return nil, err
    }
    ap := &AsyncProducer{producer: p}
    ap.handleResults()
    return ap, nil
}

func (p *AsyncProducer) handleResults() {
    p.wg.Add(2)

    // 处理成功
    go func() {
        defer p.wg.Done()
        for msg := range p.producer.Successes() {
            log.Printf("sent: topic=%s partition=%d offset=%d",
                msg.Topic, msg.Partition, msg.Offset)
        }
    }()

    // 处理失败（可加入重试/告警逻辑）
    go func() {
        defer p.wg.Done()
        for err := range p.producer.Errors() {
            log.Printf("failed: %v", err)
            // TODO: 写入失败消息到本地文件或告警
        }
    }()
}

// Input 发送消息（非阻塞）
func (p *AsyncProducer) Input(msg *sarama.ProducerMessage) {
    p.producer.Input() <- msg
}

func (p *AsyncProducer) Close() error {
    p.producer.AsyncClose()
    p.wg.Wait()
    return nil
}

// pkg/kafka/consumer.go
package kafka

import (
    "context"
    "fmt"
    "log"
    "sync"

    "github.com/IBM/sarama"
)

// HandlerFunc 消息处理函数，返回 error 时消息不提交（等待重试）
type HandlerFunc func(ctx context.Context, msg *sarama.ConsumerMessage) error

type ConsumerGroup struct {
    group   sarama.ConsumerGroup
    topics  []string
    handler HandlerFunc
}

func NewConsumerGroup(brokers []string, groupID string, topics []string,
    cfg *sarama.Config, handler HandlerFunc) (*ConsumerGroup, error) {

    g, err := sarama.NewConsumerGroup(brokers, groupID, cfg)
    if err != nil {
        return nil, fmt.Errorf("new consumer group: %w", err)
    }
    return &ConsumerGroup{group: g, topics: topics, handler: handler}, nil
}

// Start 启动消费（阻塞，直到 ctx 取消）
func (c *ConsumerGroup) Start(ctx context.Context) error {
    h := &consumerGroupHandler{handler: c.handler}

    for {
        // 每次 Rebalance 后重新进入消费循环
        if err := c.group.Consume(ctx, c.topics, h); err != nil {
            return fmt.Errorf("consume: %w", err)
        }
        if ctx.Err() != nil {
            return ctx.Err()
        }
        log.Println("rebalance completed, resuming...")
    }
}

func (c *ConsumerGroup) Close() error {
    return c.group.Close()
}

// consumerGroupHandler 实现 sarama.ConsumerGroupHandler 接口
type consumerGroupHandler struct {
    handler HandlerFunc
    wg      sync.WaitGroup
}

func (h *consumerGroupHandler) Setup(sess sarama.ConsumerGroupSession) error {
    log.Printf("consumer group setup: member=%s generation=%d",
        sess.MemberID(), sess.GenerationID())
    return nil
}

func (h *consumerGroupHandler) Cleanup(sess sarama.ConsumerGroupSession) error {
    log.Printf("consumer group cleanup, waiting for in-flight messages...")
    h.wg.Wait()
    return nil
}

func (h *consumerGroupHandler) ConsumeClaim(
    sess sarama.ConsumerGroupSession,
    claim sarama.ConsumerGroupClaim,
) error {
    for msg := range claim.Messages() {
        h.wg.Add(1)
        // 串行处理（保证分区内有序）
        func() {
            defer h.wg.Done()
            ctx := sess.Context()
            if err := h.handler(ctx, msg); err != nil {
                log.Printf("handler error: %v, topic=%s partition=%d offset=%d",
                    err, msg.Topic, msg.Partition, msg.Offset)
                // 不提交 offset，等待重试
                return
            }
            // 标记消息，下次 CommitOffsets 时提交
            sess.MarkMessage(msg, "")
            sess.CommitOffsets()
        }()
    }
    return nil
}

// examples/order_events/main.go
package main

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "os"
    "os/signal"
    "syscall"
    "time"

    "github.com/IBM/sarama"
    "github.com/your/project/pkg/kafka"
)

const (
    brokerList = "localhost:9092"
    topic      = "order.created"
)

type OrderCreatedEvent struct {
    OrderID   string    `json:"order_id"`
    UserID    int64     `json:"user_id"`
    Items     []Item    `json:"items"`
    Amount    float64   `json:"amount"`
    CreatedAt time.Time `json:"created_at"`
}

type Item struct {
    SkuID    string `json:"sku_id"`
    Quantity int    `json:"quantity"`
    Price    float64 `json:"price"`
}

// ========== 生产者：订单服务 ==========

func createOrder(producer *kafka.SyncProducer, order OrderCreatedEvent) error {
    body, err := json.Marshal(order)
    if err != nil {
        return err
    }
    // 以 OrderID 为 Key，保证同一订单的消息进同一分区（顺序保证）
    partition, offset, err := producer.Send(topic, order.OrderID, body)
    if err != nil {
        return fmt.Errorf("publish order event: %w", err)
    }
    log.Printf("order event sent: order_id=%s partition=%d offset=%d",
        order.OrderID, partition, offset)
    return nil
}

// ========== 消费者：库存服务 ==========

func inventoryHandler(ctx context.Context, msg *sarama.ConsumerMessage) error {
    var event OrderCreatedEvent
    if err := json.Unmarshal(msg.Value, &event); err != nil {
        return err
    }
    log.Printf("[库存服务] 扣减库存: order_id=%s", event.OrderID)
    for _, item := range event.Items {
        log.Printf("  sku=%s qty=%d", item.SkuID, item.Quantity)
        // deductInventory(item.SkuID, item.Quantity)
    }
    return nil
}

// ========== 消费者：积分服务 ==========

func rewardHandler(ctx context.Context, msg *sarama.ConsumerMessage) error {
    var event OrderCreatedEvent
    if err := json.Unmarshal(msg.Value, &event); err != nil {
        return err
    }
    points := int(event.Amount)
    log.Printf("[积分服务] 用户 %d 获得 %d 积分", event.UserID, points)
    // addUserPoints(event.UserID, points)
    return nil
}

func main() {
    brokers := []string{brokerList}
    cfg := kafka.Config{Brokers: brokers, Version: "3.6.0", ClientID: "order-demo"}

    // 启动生产者
    go func() {
        prodCfg := kafka.NewProducerConfig(cfg)
        producer, err := kafka.NewSyncProducer(brokers, prodCfg)
        if err != nil {
            log.Fatal(err)
        }
        defer producer.Close()

        for i := 1; i <= 10; i++ {
            order := OrderCreatedEvent{
                OrderID:   fmt.Sprintf("ORD-%06d", i),
                UserID:    int64(1000 + i),
                Amount:    float64(i) * 99.9,
                CreatedAt: time.Now(),
                Items: []Item{
                    {SkuID: fmt.Sprintf("SKU-%04d", i), Quantity: 1, Price: float64(i) * 99.9},
                },
            }
            createOrder(producer, order)
            time.Sleep(200 * time.Millisecond)
        }
    }()

    ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
    defer stop()

    consCfg := kafka.NewConsumerConfig(cfg)

    // 库存服务消费者组
    inventoryGroup, _ := kafka.NewConsumerGroup(
        brokers, "inventory-service", []string{topic}, consCfg, inventoryHandler)
    defer inventoryGroup.Close()

    // 积分服务消费者组（独立消费，互不影响）
    rewardGroup, _ := kafka.NewConsumerGroup(
        brokers, "reward-service", []string{topic}, consCfg, rewardHandler)
    defer rewardGroup.Close()

    go inventoryGroup.Start(ctx)
    go rewardGroup.Start(ctx)

    <-ctx.Done()
    log.Println("shutting down...")
}

// 重置到指定时间点重新消费（数据修复场景）
func resetOffsetToTime(brokers []string, groupID, topic string, t time.Time) error {
    cfg := sarama.NewConfig()
    cfg.Version = sarama.V3_6_0_0

    admin, err := sarama.NewClusterAdmin(brokers, cfg)
    if err != nil {
        return err
    }
    defer admin.Close()

    client, err := sarama.NewClient(brokers, cfg)
    if err != nil {
        return err
    }
    defer client.Close()

    partitions, _ := client.Partitions(topic)

    offsets := make(map[string]map[int32]int64)
    offsets[topic] = make(map[int32]int64)

    for _, partition := range partitions {
        // 获取指定时间的 offset
        offset, err := client.GetOffset(topic, partition, t.UnixMilli())
        if err != nil {
            return err
        }
        offsets[topic][partition] = offset
        log.Printf("partition=%d reset to offset=%d (time=%s)", partition, offset, t)
    }

    return admin.ResetConsumerGroupOffsets(groupID, offsets)
}

package main

import (
    "context"
    "encoding/json"
    "log"
    "time"

    "github.com/IBM/sarama"
)

type LogEntry struct {
    Service   string    `json:"service"`
    Level     string    `json:"level"`
    Message   string    `json:"message"`
    TraceID   string    `json:"trace_id"`
    Timestamp time.Time `json:"timestamp"`
}

// 生产者：各服务打日志
func writeLog(producer sarama.AsyncProducer, entry LogEntry) {
    body, _ := json.Marshal(entry)
    producer.Input() <- &sarama.ProducerMessage{
        Topic: "app.logs",
        Key:   sarama.StringEncoder(entry.Service),
        Value: sarama.ByteEncoder(body),
    }
}

// 消费者：批量写入 ES
type ESBatcher struct {
    batch   []*LogEntry
    maxSize int
    timeout time.Duration
    ticker  *time.Ticker
}

func NewESBatcher(maxSize int, timeout time.Duration) *ESBatcher {
    return &ESBatcher{
        batch:   make([]*LogEntry, 0, maxSize),
        maxSize: maxSize,
        timeout: timeout,
        ticker:  time.NewTicker(timeout),
    }
}

func (b *ESBatcher) Process(ctx context.Context, msg *sarama.ConsumerMessage) error {
    var entry LogEntry
    if err := json.Unmarshal(msg.Value, &entry); err != nil {
        return err
    }
    b.batch = append(b.batch, &entry)

    select {
    case <-b.ticker.C:
        return b.flush()
    default:
        if len(b.batch) >= b.maxSize {
            return b.flush()
        }
    }
    return nil
}

func (b *ESBatcher) flush() error {
    if len(b.batch) == 0 {
        return nil
    }
    log.Printf("flushing %d logs to ES", len(b.batch))
    // bulkInsertToES(b.batch)
    b.batch = b.batch[:0]
    return nil
}

// 使用 Kafka 事务实现跨 Topic 的 Exactly Once
// 场景：从 source-topic 消费，处理后写入 dest-topic，保证原子性

func processWithExactlyOnce(
    client sarama.Client,
    srcTopic, dstTopic string,
    groupID string,
) error {
    // 事务生产者
    txnCfg := sarama.NewConfig()
    txnCfg.Producer.Transaction.ID = "eos-processor-1"
    txnCfg.Producer.RequiredAcks = sarama.WaitForAll
    txnCfg.Producer.Idempotent = true
    txnCfg.Net.MaxOpenRequests = 1

    producer, err := sarama.NewSyncProducer(client.Brokers(), txnCfg)
    if err != nil {
        return err
    }
    defer producer.Close()

    // 普通消费者（offset 通过事务提交，不使用 consumer group commit）
    consumer, _ := sarama.NewConsumerFromClient(client)
    pc, _ := consumer.ConsumePartition(srcTopic, 0, sarama.OffsetNewest)

    for msg := range pc.Messages() {
        // 开始事务
        producer.(sarama.SyncProducer)

        // 处理消息
        result := process(msg.Value)

        // 在事务中提交 offset 和写入目标 Topic
        // kafka 事务确保这两个操作原子执行
        offsets := map[string][]*sarama.PartitionOffsetMetadata{
            srcTopic: {{Partition: msg.Partition, Offset: msg.Offset + 1}},
        }
        _ = offsets // 实际使用 producer.AddOffsetsToTxn(offsets, groupID)

        producer.SendMessages([]*sarama.ProducerMessage{
            {Topic: dstTopic, Value: sarama.ByteEncoder(result)},
        })
    }
    return nil
}

func process(data []byte) []byte {
    // 业务处理逻辑
    return data
}

# docker-compose.yml
version: '3.8'

x-kafka-common: &kafka-common
  image: apache/kafka:3.7.0
  environment: &kafka-env
    KAFKA_PROCESS_ROLES: broker,controller
    KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka1:9093,2@kafka2:9093,3@kafka3:9093
    KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
    KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
    KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
    KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT
    KAFKA_DEFAULT_REPLICATION_FACTOR: "3"
    KAFKA_MIN_INSYNC_REPLICAS: "2"
    KAFKA_NUM_PARTITIONS: "6"
    KAFKA_LOG_RETENTION_HOURS: "168"
    KAFKA_LOG_SEGMENT_BYTES: "1073741824"
    CLUSTER_ID: "MkU3OEVBNTcwNTJENDM2Qk"  # 需固定，使用 kafka-storage random-uuid 生成

services:
  kafka1:
    <<: *kafka-common
    hostname: kafka1
    environment:
      <<: *kafka-env
      KAFKA_NODE_ID: "1"
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
    ports:
      - "9092:9092"
    volumes:
      - kafka1_data:/var/lib/kafka/data

  kafka2:
    <<: *kafka-common
    hostname: kafka2
    environment:
      <<: *kafka-env
      KAFKA_NODE_ID: "2"
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9093
    ports:
      - "9093:9092"
    volumes:
      - kafka2_data:/var/lib/kafka/data

  kafka3:
    <<: *kafka-common
    hostname: kafka3
    environment:
      <<: *kafka-env
      KAFKA_NODE_ID: "3"
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9094
    ports:
      - "9094:9092"
    volumes:
      - kafka3_data:/var/lib/kafka/data

  kafka-ui:
    image: provectuslabs/kafka-ui:latest
    ports:
      - "8080:8080"
    environment:
      KAFKA_CLUSTERS_0_NAME: local
      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka1:9092,kafka2:9092,kafka3:9092

volumes:
  kafka1_data:
  kafka2_data:
  kafka3_data:

# 创建 Topic（6个分区，3个副本）
kafka-topics.sh --bootstrap-server localhost:9092 \
    --create --topic orders \
    --partitions 6 \
    --replication-factor 3 \
    --config retention.ms=604800000 \
    --config min.insync.replicas=2

# 查看 Topic 详情
kafka-topics.sh --bootstrap-server localhost:9092 \
    --describe --topic orders

# 修改分区数（只能增加）
kafka-topics.sh --bootstrap-server localhost:9092 \
    --alter --topic orders --partitions 12

# 删除 Topic
kafka-topics.sh --bootstrap-server localhost:9092 \
    --delete --topic orders

# 列出所有 Topic
kafka-topics.sh --bootstrap-server localhost:9092 --list

# 生产消息（命令行）
kafka-console-producer.sh --bootstrap-server localhost:9092 \
    --topic orders \
    --property "key.separator=:" \
    --property "parse.key=true"
# 输入: order-001:{"amount":100}

# 消费消息（从头开始）
kafka-console-consumer.sh --bootstrap-server localhost:9092 \
    --topic orders \
    --from-beginning \
    --property print.key=true \
    --property print.partition=true \
    --property print.offset=true

# 查看消费组 Lag
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
    --group inventory-service --describe

# 重置消费 offset 到最早
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
    --group inventory-service \
    --topic orders \
    --reset-offsets --to-earliest \
    --execute

# kafka-server-start.sh
export KAFKA_HEAP_OPTS="-Xms6g -Xmx6g"
export KAFKA_JVM_PERFORMANCE_OPTS="-XX:+UseG1GC \
    -XX:MaxGCPauseMillis=20 \
    -XX:InitiatingHeapOccupancyPercent=35 \
    -XX:G1HeapRegionSize=16M \
    -XX:MinMetaspaceFreeRatio=50 \
    -XX:MaxMetaspaceFreeRatio=80"

# 增大文件句柄数（每个分区需要多个文件句柄）
echo "* soft nofile 1000000" >> /etc/security/limits.conf
echo "* hard nofile 1000000" >> /etc/security/limits.conf

# 调整内核参数
sysctl -w vm.swappiness=1          # 几乎不用 swap
sysctl -w net.core.rmem_max=134217728
sysctl -w net.core.wmem_max=134217728
sysctl -w net.ipv4.tcp_rmem="4096 131072 134217728"
sysctl -w net.ipv4.tcp_wmem="4096 131072 134217728"

// 追求吞吐量
config.Producer.Flush.Bytes = 5 * 1024 * 1024    // 5MB 批次
config.Producer.Flush.Frequency = time.Second      // 或1秒发一次
config.Producer.Compression = sarama.CompressionLZ4 // LZ4 压缩，速度最快

// 追求低延迟
config.Producer.Flush.Bytes = 1                    // 立即发送
config.Producer.Flush.Frequency = time.Millisecond
config.Producer.Compression = sarama.CompressionNone

// 增大每次 fetch 的数据量，减少 fetch 次数
config.Consumer.Fetch.Min = 1024 * 1024     // 最小 fetch 1MB
config.Consumer.Fetch.Default = 10 * 1024 * 1024 // 默认 fetch 10MB
config.Consumer.MaxWaitTime = 500 * time.Millisecond // 最长等待 500ms

func handleMessage(ctx context.Context, msg *sarama.ConsumerMessage) error {
    // 从消息中提取业务唯一ID（或使用 topic+partition+offset 作为唯一ID）
    var payload struct {
        OrderID string `json:"order_id"`
    }
    json.Unmarshal(msg.Value, &payload)

    // 方案1：Redis 去重
    key := fmt.Sprintf("kafka:processed:%s:%d:%d",
        msg.Topic, msg.Partition, msg.Offset)
    ok, _ := redis.SetNX(ctx, key, 1, 24*time.Hour)
    if !ok {
        return nil // 已处理，直接跳过
    }

    // 方案2：数据库唯一索引
    // INSERT IGNORE INTO processed_messages (topic, partition, offset) VALUES (?,?,?)

    return processOrder(payload.OrderID)
}

// 静态成员（固定 instance ID，重启不触发 Rebalance）
config.Consumer.Group.InstanceID = "consumer-instance-1"

# 1. 查看积压量
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
    --group my-group --describe

# 2. 查看 Consumer 处理速度（Consume Rate vs Produce Rate）
# 通过 Prometheus/JMX 看 records-consumed-rate vs records-produced-rate

// 并发消费同一分区的消息（牺牲顺序换吞吐）
func (h *handler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
    sem := make(chan struct{}, 20) // 20并发
    var wg sync.WaitGroup
    for msg := range claim.Messages() {
        sem <- struct{}{}
        wg.Add(1)
        go func(m *sarama.ConsumerMessage) {
            defer func() { <-sem; wg.Done() }()
            h.process(m)
            sess.MarkMessage(m, "")
        }(msg)
    }
    wg.Wait()
    return nil
}