Golang微服务消息通知与事件驱动实践

github.com/segmentio/kafka-go

github.com/confluentinc/confluent-kafka-go

package main

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

    "github.com/segmentio/kafka-go"
)

type OrderCreatedEvent struct {
    OrderID string  `json:"order_id"`
    UserID  string  `json:"user_id"`
    Amount  float64 `json:"amount"`
    // ... 其他订单详情
}

var kafkaWriter *kafka.Writer // 假设这是一个已初始化的Kafka生产者

func init() {
    // 实际应用中，这里会根据配置初始化Kafka writer
    // 示例中简化，假设已配置好
    kafkaWriter = &kafka.Writer{
        Addr:     kafka.TCP("localhost:9092"), // 替换为你的Kafka地址
        Topic:    "order_events",
        Balancer: &kafka.LeastBytes{},
    }
}

func publishOrderCreated(event OrderCreatedEvent) error {
    messageBytes, err := json.Marshal(event)
    if err != nil {
        log.Printf("Error marshalling event: %v", err)
        return err
    }

    err = kafkaWriter.WriteMessages(context.Background(),
        kafka.Message{
            Key:   []byte(event.OrderID), // 通常用业务ID作为Key，确保相关消息进入同一分区
            Value: messageBytes,
        },
    )
    if err != nil {
        log.Printf("Failed to publish order created event: %v", err)
        return err
    }
    log.Printf("Published OrderCreatedEvent for OrderID: %s", event.OrderID)
    return nil
}

func main() {
    // 模拟订单创建并发布事件
    event := OrderCreatedEvent{
        OrderID: "ORD12345",
        UserID:  "USR001",
        Amount:  99.99,
    }
    if err := publishOrderCreated(event); err != nil {
        log.Fatalf("Failed to publish event: %v", err)
    }

    // 在实际应用中，这里不会直接关闭writer，而是由服务生命周期管理
    // defer kafkaWriter.Close()
}

package main

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

    "github.com/segmentio/kafka-go"
)

type OrderCreatedEvent struct {
    OrderID string  `json:"order_id"`
    UserID  string  `json:"user_id"`
    Amount  float64 `json:"amount"`
    // ... 其他订单详情
}

func consumeOrderEvents() {
    // 实际应用中，这里会根据配置初始化Kafka reader
    r := kafka.NewReader(kafka.ReaderConfig{
        Brokers:        []string{"localhost:9092"}, // 替换为你的Kafka地址
        Topic:          "order_events",
        GroupID:        "payment-service-group", // 消费者组ID，确保消息只被组内一个实例消费
        MinBytes:       10e3,                    // 10KB
        MaxBytes:       10e6,                    // 10MB
        CommitInterval: time.Second,             // 每秒提交一次偏移量
        // ReadBackoffMin: time.Millisecond * 100, // 消费失败重试间隔
        // ReadBackoffMax: time.Second * 5,
    })

    defer r.Close()

    log.Println("Payment service started consuming order_events...")
    for {
        m, err := r.ReadMessage(context.Background())
        if err != nil {
            log.Printf("Error reading message: %v", err)
            // 考虑错误处理，如短暂网络问题可重试，严重错误记录日志或退出
            time.Sleep(time.Second * 5) // 简单重试间隔
            continue
        }

        var event OrderCreatedEvent
        if err := json.Unmarshal(m.Value, &event); err != nil {
            log.Printf("Error unmarshalling event from partition %d, offset %d: %v", m.Partition, m.Offset, err)
            // 消息格式错误，通常会记录到死信队列 (DLQ)
            continue
        }

        log.Printf("Received OrderCreatedEvent from partition %d, offset %d for OrderID: %s, UserID: %s, Amount: %.2f",
            m.Partition, m.Offset, event.OrderID, event.UserID, event.Amount)

        // --- 执行支付相关逻辑 ---
        // 1. 检查幂等性：确保该订单ID的支付操作未重复执行
        //    例如：查询支付记录，如果已存在，则跳过
        // 2. 调用支付网关或更新本地支付状态
        // 3. 如果支付成功，可能发布新的支付成功事件
        // --- 支付逻辑结束 ---

        // 显式提交偏移量，表示消息已成功处理
        // r.CommitMessages(context.Background(), m) // ReadMessage会自动提交，但手动控制更精细
        log.Printf("Successfully processed OrderID: %s", event.OrderID)
    }
}

func main() {
    consumeOrderEvents()
}