Go消息队列选型与Docker部署实战：NATS、Asynq、RabbitMQ对比

1
package main
2

3
import (
4
  "log"
5
  "time"
6
  "github.com/nats-io/nats.go"
7
)
8

9
func main() {
10
  nc, _ := nats.Connect("nats://localhost:4222")
11
  js, _ := nc.JetStream()
12
  defer nc.Close()
13

14
  // 1. 创建 Stream（持久化消息流）
15
  js.AddStream(&nats.StreamConfig{
16
    Name:        "ORDERS",
17
    Description: "订单事件流",
18
    Subjects:    []string{"orders.>"},
19
    Storage:     nats.FileStorage,
20
    MaxAge:      7 * 24 * time.Hour, // 保留7天
21
  })
22

23
  // 2. 为不同消费者创建不同的 Consumer
24
  // 仓储服务 —— 只关心 orders.created
25
  js.AddConsumer("ORDERS", &nats.ConsumerConfig{
26
    Name:          "warehouse",
27
    FilterSubject: "orders.created",
28
    AckPolicy:     nats.AckExplicitPolicy,
29
  })
30

31
  // 邮件服务 —— 也关心 orders.created
32
  js.AddConsumer("ORDERS", &nats.ConsumerConfig{
33
    Name:          "email",
34
    FilterSubject: "orders.created",
35
    AckPolicy:     nats.AckExplicitPolicy,
36
  })
37

38
  // 日志服务 —— 关心所有订单事件
39
  js.AddConsumer("ORDERS", &nats.ConsumerConfig{
40
    Name:          "logger",
41
    FilterSubject: "orders.>",
42
    AckPolicy:     nats.AckExplicitPolicy,
43
  })
44

45
  // 3. 发布订单事件
46
  js.Publish("orders.created", []byte(`{"order_id":"ORD-001","user_id":42,"amount":199.0}`))
47
  log.Println("订单事件已发布")
48

49
  // 4. 仓储服务消费
50
  sub, _ := js.PullSubscribe("orders.created", "warehouse")
51
  msgs, _ := sub.Fetch(10, nats.MaxWait(2*time.Second))
52
  for _, msg := range msgs {
53
    log.Printf("仓储服务收到: %s", string(msg.Data))
54
    msg.AckSync()
55
  }
56
}

1
package main
2

3
import (
4
  "log"
5
  "time"
6
  "github.com/hibiken/asynq"
7
)
8

9
const TypeDailyReport = "report:daily"
10

11
func main() {
12
  client := asynq.NewClient(asynq.RedisClientOpt{Addr: "localhost:6379"})
13
  defer client.Close()
14

15
  // 注册定时任务 —— 每天早上 8 点执行
16
  scheduler := asynq.NewScheduler(
17
    asynq.RedisClientOpt{Addr: "localhost:6379"},
18
    &asynq.SchedulerOpts{Location: time.Local},
19
  )
20

21
  task := asynq.NewTask(TypeDailyReport, []byte(`{"type": "daily_summary"}`))
22
  // CRON 表达式：每天 8:00 执行
23
  entryID, _ := scheduler.Register("@daily 08:00", task)
24
  log.Printf("定时任务已注册: %s", entryID)
25

26
  // 启动调度器
27
  scheduler.Run()
28

29
  // 消费者端处理
30
  srv := asynq.NewServer(
31
    asynq.RedisClientOpt{Addr: "localhost:6379"},
32
    asynq.Config{Concurrency: 10},
33
  )
34
  mux := asynq.NewServeMux()
35
  mux.HandleFunc(TypeDailyReport, func(ctx context.Context, t *asynq.Task) error {
36
    log.Printf("生成日报并推送")
37
    return nil
38
  })
39
  srv.Run(mux)
40
}