点赞收藏与 Kafka 入门

-- interactive 表：互动计数表
CREATE TABLE `interactive` (
  `id` bigint UNSIGNED NOT NULL AUTO_INCREMENT,
  `biz` varchar(32) NOT NULL COMMENT '业务标识，如 article',
  `biz_id` bigint NOT NULL COMMENT '业务对象 ID',
  `read_cnt` bigint NOT NULL DEFAULT 0,
  `like_cnt` bigint NOT NULL DEFAULT 0,
  `collect_cnt` bigint NOT NULL DEFAULT 0,
  `ctime` bigint NOT NULL,
  `utime` bigint NOT NULL,
  PRIMARY KEY (`id`),
  -- 关键：联合唯一索引，保证同一 biz+bizId 只有一条记录
  UNIQUE KEY `biz_biz_id` (`biz`, `biz_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

// dao/interactive.go
func (d *InteractiveDAO) InsertOrUpdate(ctx context.Context, biz string, bizId int64) error {
    // Upsert：插入如果冲突则更新 read_cnt = read_cnt + 1
    // 不能 "先 SELECT 再 +1 再 UPDATE"，会有并发问题
    // 直接利用 MySQL 的原子自增
    return d.db.WithContext(ctx).Clauses(clause.OnConflict{
        Columns: []clause.Column{{Name: "biz"}, {Name: "biz_id"}},
        DoUpdates: clause.Assignments(map[string]interface{}{
            "read_cnt": gorm.Expr("read_cnt + ?", 1),
        }),
    }).Create(&Interactive{
        Biz: biz, BizId: bizId, ReadCnt: 1,
        Ctime: time.Now().UnixMilli(), Utime: time.Now().UnixMilli(),
    }).Error
}

// lua 脚本：key 存在时把指定 field 自增 delta
const luaIncrIfPresent = `
if redis.call("EXISTS", KEYS[1]) == 1 then
    return redis.call("HINCRBY", KEYS[1], ARGV[1], ARGV[2])
else
    return 0
end
`

// IncrReadCnt 的缓存操作
func (r *InteractiveRedisCache) IncrReadCntIfPresent(ctx context.Context, biz string, bizId int64) error {
    key := r.key(biz, bizId)
    // ARGV[1] = field 名，ARGV[2] = 增量
    _, err := r.client.Eval(ctx, luaIncrIfPresent, []string{key}, "read_cnt", 1).Result()
    return err
}

-- user_like biz 表：记录哪个用户对哪个资源点过赞
CREATE TABLE `user_like_biz` (
  `id` bigint UNSIGNED NOT NULL AUTO_INCREMENT,
  `biz` varchar(32) NOT NULL,
  `biz_id` bigint NOT NULL,
  `user_id` bigint NOT NULL,
  `status` tinyint NOT NULL DEFAULT 1 COMMENT '1 点赞 0 取消',
  `ctime` bigint NOT NULL,
  `utime` bigint NOT NULL,
  PRIMARY KEY (`id`),
  UNIQUE KEY `biz_biz_id_user_id` (`biz`, `biz_id`, `user_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

// biz/interactive.go
func (s *interactiveService) Like(ctx context.Context, biz string, id int64, uid int64) error {
    // 步骤 1：给资源的总点赞数 +1
    err := s.repo.IncrLikeCnt(ctx, biz, id, 1)
    if err != nil {
        return err
    }
    return s.repo.MarkLiked(ctx, biz, id, uid, true)
}

func (s *interactiveService) CancelLike(ctx context.Context, biz string, id int64, uid int64) error {
    err := s.repo.IncrLikeCnt(ctx, biz, id, -1)
    if err != nil {
        return err
    }
    return s.repo.MarkLiked(ctx, biz, id, uid, false)
}

func (d *InteractiveDAO) UpsertLike(ctx context.Context, biz string, bizId, uid int64, status int8) error {
    var st int8
    if status { st = 1 } else { st = 0 }
    return d.db.WithContext(ctx).Clauses(clause.OnConflict{
        Columns: []clause.Column{{Name: "biz"}, {Name: "biz_id"}, {Name: "user_id"}},
        DoUpdates: clause.Assignments(map[string]interface{}{
            "status": st,
            "utime":  time.Now().UnixMilli(),
        }),
    }).Create(&UserLikeBiz{
        Biz: biz, BizId: bizId, UserId: uid, Status: st,
        Ctime: time.Now().UnixMilli(), Utime: time.Now().UnixMilli(),
    }).Error
}

-- 收藏夹本体
CREATE TABLE `collection` (
  `id` bigint UNSIGNED NOT NULL AUTO_INCREMENT,
  `name` varchar(128) NOT NULL,
  `user_id` bigint NOT NULL,
  `ctime` bigint NOT NULL,
  `utime` bigint NOT NULL,
  PRIMARY KEY (`id`),
  KEY `idx_user_id` (`user_id`)
);

-- 收藏夹与资源的关联
CREATE TABLE `collection_item` (
  `id` bigint UNSIGNED NOT NULL AUTO_INCREMENT,
  `collection_id` bigint NOT NULL,
  `biz` varchar(32) NOT NULL,
  `biz_id` bigint NOT NULL,
  `user_id` bigint NOT NULL,
  `ctime` bigint NOT NULL,
  `utime` bigint NOT NULL,
  PRIMARY KEY (`id`),
  UNIQUE KEY `biz_biz_id_user_id` (`biz`, `biz_id`, `user_id`)
);

func (d *InteractiveDAO) AddCollectionItem(ctx context.Context, biz string, bizId, uid, colId int64) error {
    return d.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
        // 1. 插入收藏项（Upsert 防重）
        err := tx.Clauses(clause.OnConflict{
            Columns: []clause.Column{{Name: "biz"}, {Name: "biz_id"}, {Name: "user_id"}},
            DoUpdates: clause.Assignments(map[string]interface{}{
                "collection_id": colId,
                "utime":         time.Now().UnixMilli(),
            }),
        }).Create(&CollectionItem{
            CollectionId: colId, Biz: biz, BizId: bizId, UserId: uid,
            Ctime: time.Now().UnixMilli(), Utime: time.Now().UnixMilli(),
        }).Error
        if err != nil {
            return err
        }
        // 2. 更新 collect_cnt（Upsert）
        return tx.Clauses(clause.OnConflict{
            Columns: []clause.Column{{Name: "biz"}, {Name: "biz_id"}},
            DoUpdates: clause.Assignments(map[string]interface{}{
                "collect_cnt": gorm.Expr("collect_cnt + ?", 1),
            }),
        }).Create(&Interactive{
            Biz: biz, BizId: bizId, CollectCnt: 1,
            Ctime: time.Now().UnixMilli(), Utime: time.Now().UnixMilli(),
        }).Error
    })
}

func (h *ArticleHandler) PubDetail(c *gin.Context) {
    var (
        art domain.Article
        intr domain.Interactive
    )
    g, ctx := errgroup.WithContext(c.Request.Context())
    // 拉文章
    g.Go(func() error {
        var err error
        art, err = h.articleSvc.PublicById(ctx, id)
        return err
    })
    // 拉互动信息
    g.Go(func() error {
        var err error
        intr, err = h.intrSvc.Get(ctx, "article", id, uid)
        return err
    })
    if err := g.Wait(); err != nil { /* ... */ }
    // 聚合返回
}

config := sarama.NewConfig()
config.Producer.Partitioner = sarama.NewHashPartitioner // 按 key 哈希
// 发送时设置 key，比如 biz_id
msg := &sarama.ProducerMessage{
    Topic: "read_article",
    Key:   sarama.StringEncoder(strconv.FormatInt(bizId, 10)),
    Value: sarama.ByteEncoder(payload),
}

# docker-compose.yaml
services:
  kafka:
    image: bitnami/kafka:latest
    ports:
      - "9092:9092"
    environment:
      - KAFKA_CFG_NODE_ID=0
      - KAFKA_CFG_PROCESS_ROLES=controller,broker
      - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092
      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092
      - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=0@kafka:9093
      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER

func NewSyncProducer(addrs []string) (sarama.SyncProducer, error) {
    config := sarama.NewConfig()
    // acks 选择：-1 最安全，1 平衡，0 最快
    config.Producer.RequiredAcks = sarama.WaitForAll // 等价 acks=-1
    config.Producer.Return.Successes = true           // 同步发送必须开启
    return sarama.NewSyncProducer(addrs, config)
}

func produce(p sarama.SyncProducer, topic, key, value string) error {
    msg := &sarama.ProducerMessage{
        Topic: topic,
        Key:   sarama.StringEncoder(key), // 同 key 进同分区，保证业务有序
        Value: sarama.ByteEncoder(value),
    }
    _, _, err := p.SendMessage(msg)
    return err
}

func NewAsyncProducer(addrs []string) (sarama.AsyncProducer, error) {
    config := sarama.NewConfig()
    config.Producer.Return.Successes = true // 想拿到成功通知就开启
    config.Producer.Return.Errors = true    // 想拿到错误通知就开启
    return sarama.NewAsyncProducer(addrs, config)
}

func runAsyncProducer(p sarama.AsyncProducer) {
    // 发送消息：直接丢进 Input channel
    p.Input() <- &sarama.ProducerMessage{Topic: "test", Value: sarama.StringEncoder("hi")}

    // 用 select 同时监听成功和失败
    for {
        select {
        case msg := <-p.Successes():
            log.Printf("发送成功 partition=%d offset=%d", msg.Partition, msg.Offset)
        case err := <-p.Errors():
            log.Printf("发送失败 %v", err)
        }
    }
}

type readArticleHandler struct {
    // 注入业务 service
}

// Setup 在消费者组 rebalance 之后、消费之前调用
func (h *readArticleHandler) Setup(sarama.ConsumerGroupSession) error { return nil }

// Cleanup 在消费结束、rebalance 之前调用
func (h *readArticleHandler) Cleanup(sarama.ConsumerGroupSession) error { return nil }

// ConsumeClaim 核心消费逻辑
func (h *readArticleHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
    for msg := range claim.Messages() {
        // 1. 业务处理
        err := h.svc.OnReadEvent(context.Background(), msg.Value)
        if err != nil {
            // 失败处理：重试 / 死信队列 / 记录日志后跳过
            log.Printf("消费失败 topic=%s partition=%d offset=%d err=%v",
                msg.Topic, msg.Partition, msg.Offset, err)
        }
        // 2. 业务成功后标记消费完成，提交偏移量
        sess.MarkMessage(msg, "")
    }
    return nil
}

ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go func() {
    for {
        // Consume 阻塞消费，出错或 ctx 取消会返回
        err := cg.Consume(ctx, []string{"read_article"}, &readArticleHandler{})
        if err != nil { log.Println(err) }
        if ctx.Err() != nil { return } // 主动退出
    }
}()

// 领域事件定义
type ReadArticleEvent struct {
    Uid int64
    Aid int64
}

// ArticleService 内部发送消息
func (s *articleService) PublicById(ctx context.Context, id, uid int64) (domain.Article, error) {
    art, err := s.repo.PublicById(ctx, id)
    if err != nil { return art, err }
    // 异步发送领域事件，不阻塞主流程
    evt := ReadArticleEvent{Uid: uid, Aid: id}
    payload, _ := json.Marshal(evt)
    _ = s.producer.Produce(ctx, "read_article", strconv.FormatInt(id, 10), payload)
    return art, nil
}

func (h *readArticleHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
    for msg := range claim.Messages() {
        var evt ReadArticleEvent
        _ = json.Unmarshal(msg.Value, &evt)
        _ = h.intrSvc.IncrReadCnt(context.Background(), "article", evt.Aid)
        // 步骤：业务成功后提交偏移量，否则重启会重复消费
        sess.MarkMessage(msg, "")
    }
    return nil
}

> ⚠️ **新手必踩的坑：忘记 `MarkMessage` 会重复消费**。只有调用 `MarkMessage` 提交偏移量，Kafka 才认为这条消息"消费过了"。如果处理成功却没提交，消费者重启后会从旧偏移量重新拉，这条消息被再处理一次。所以"先确保业务成功，再 MarkMessage"是铁律。

type batchReadHandler struct {
    svc   InteractiveService
    batch int // 一批大小，比如 100
    wait  time.Duration // 凑批超时
}

func (h *batchReadHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
    batch := make([]int64, 0, h.batch)
    timer := time.NewTimer(h.wait)
    defer timer.Stop()
    for {
        select {
        case msg, ok := <-claim.Messages():
            if !ok { return nil }
            var evt ReadArticleEvent
            _ = json.Unmarshal(msg.Value(), &evt)
            batch = append(batch, evt.Aid)
            sess.MarkMessage(msg, "")
            if len(batch) >= h.batch {
                h.flush(batch)
                batch = batch[:0]
                timer.Reset(h.wait)
            }
        case <-timer.C:
            if len(batch) > 0 {
                h.flush(batch)
                batch = batch[:0]
            }
            timer.Reset(h.wait)
        }
    }
}

func (h *batchReadHandler) flush(aids []int64) {
    // 调用批量接口 BatchIncrReadCnt，一次事务更新
    _ = h.svc.BatchIncrReadCnt(context.Background(), "article", aids)
}

func (d *InteractiveDAO) BatchIncrReadCnt(ctx context.Context, biz string, ids []int64) error {
    return d.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
        for _, id := range ids {
            err := tx.Clauses(clause.OnConflict{
                Columns: []clause.Column{{Name: "biz"}, {Name: "biz_id"}},
                DoUpdates: clause.Assignments(map[string]interface{}{
                    "read_cnt": gorm.Expr("read_cnt + ?", 1),
                }),
            }).Create(&Interactive{Biz: biz, BizId: id, ReadCnt: 1,
                Ctime: time.Now().UnixMilli(), Utime: time.Now().UnixMilli()}).Error
            if err != nil { return err }
        }
        return nil
    })
}

// 声明与创建
var ch chan int           // 声明
ch = make(chan int)       // 无缓冲
ch = make(chan int, 10)   // 有缓冲，容量 10

// 发送与接收
ch <- 42                  // 发送
val := <-ch               // 接收
val, ok := <-ch           // 接收并判断是否关闭

// 关闭
close(ch)
// 关闭后：写入 panic；读取得到零值，ok=false；重复 close panic

// for range 读取，close 后自动退出
for v := range ch {
    fmt.Println(v)
}

// select 多路复用
select {
case ch <- 1:
case v := <-ch2:
    fmt.Println(v)
default:
    // 所有 case 阻塞时执行
}