Go 与消息队列：邮件服务挂了，注册也不能挂

go install github.com/nats-io/nats-server/v2@latest
nats-server            # 默认监听 4222
go get github.com/nats-io/nats.go

package main

import (
    "fmt"
    "log"

    "github.com/nats-io/nats.go"
)

func main() {
    nc, err := nats.Connect("nats://localhost:4222")
    if err != nil {
        log.Fatal(err)
    }
    defer nc.Close()

    // 订阅：收到消息触发回调
    sub, err := nc.Subscribe("user.registered", func(msg *nats.Msg) {
        fmt.Printf("邮件服务收到: %s\n", string(msg.Data))
    })
    if err != nil {
        log.Fatal(err)
    }
    defer sub.Unsubscribe()

    // 发布：发完即走，不等任何订阅者
    if err := nc.Publish("user.registered", []byte(`{"email":"user@example.com"}`)); err != nil {
        log.Fatal(err)
    }

    select {} // 演示程序：保持进程存活等回调
}

邮件服务收到: {"email":"user@example.com"}

// 请求-回复：像 HTTP 但异步，Request 自带超时
nc.Subscribe("service.add", func(msg *nats.Msg) {
    msg.Respond([]byte("42"))
})
reply, err := nc.Request("service.add", []byte("1+1"), 2*time.Second)

// 队列订阅：多个实例共用一个队列组名，消息只被组内一个消费者处理
nc.QueueSubscribe("tasks", "worker-group", handler)

js, err := nc.JetStream()

// 创建流：类似 Kafka 的 Topic，带保留策略
_, err = js.AddStream(&nats.StreamConfig{
    Name:      "ORDERS",
    Subjects:  []string{"orders.*"},
    Retention: nats.LimitsPolicy,
})

// 发布持久化消息
js.Publish("orders.new", []byte(`{"id":"123"}`))

// 持久化订阅：处理完必须 Ack，不 Ack 会在重投策略下重新送达
sub, _ := js.Subscribe("orders.*", func(msg *nats.Msg) {
    if err := handle(msg.Data); err != nil {
        return // 不 Ack，等待重投
    }
    msg.Ack()
}, nats.Durable("order-processor"))

// 生产者：Produce 只是入队，结果异步到达 Events 通道
p, err := kafka.NewProducer(&kafka.ConfigMap{
    "bootstrap.servers": "localhost:9092",
    "acks":              "all", // 等 ISR 副本全部落盘才算成功，防 broker 宕机丢消息
})
defer p.Close()

go func() {
    for e := range p.Events() {
        if msg, ok := e.(*kafka.Message); ok && msg.TopicPartition.Error != nil {
            log.Printf("发送失败: %v", msg.TopicPartition.Error) // 这里接重试/告警
        }
    }
}()

topic := "orders"
p.Produce(&kafka.Message{
    TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny},
    Value:          []byte(`{"id":"123"}`),
}, nil)

p.Flush(15 * 1000) // 退出前把队列发完

c, err := kafka.NewConsumer(&kafka.ConfigMap{
    "bootstrap.servers":  "localhost:9092",
    "group.id":           "order-processor",
    "auto.offset.reset":  "earliest",        // 无偏移记录时从头消费
    "enable.auto.commit": false,             // 关闭自动提交，处理完再确认
})
defer c.Close()

c.SubscribeTopics([]string{"orders"}, nil)

for {
    msg, err := c.ReadMessage(-1) // 阻塞等待
    if err != nil {
        log.Printf("消费错误: %v", err)
        continue
    }
    if err := processMessage(msg); err != nil {
        log.Printf("处理失败: %v", err)
        continue // 不提交偏移，重启后会重新拿到这条
    }
    if _, err := c.CommitMessage(msg); err != nil {
        log.Printf("提交失败: %v", err)
    }
}

// 订单服务只管发布
js.Publish("order.created", orderData)

// 库存服务与通知服务各自订阅，互不知晓
js.Subscribe("order.created", func(msg *nats.Msg) {
    deductStock(msg.Data)
    msg.Ack()
})

sigChan := make(chan os.Signal, 1)
signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
<-sigChan
fmt.Println("正在关闭消费者...")
c.Close() // 让当前消息处理完、提交偏移、离开消费组