Stream 流队列
Stream 是不可变、仅追加的日志队列，支持消息重放与多消费者独立消费。本章覆盖 Stream 队列的声明、发布、带偏移量控制的消费，以及基于 Bloom Filter 的服务端过滤。
Stream 队列简介
Stream 是一种不可变的、仅追加的日志队列，适用于：
- 大吞吐量的消息流
- 消息重放
- 多消费者独立消费（非竞争模式）
- 事件溯源
声明 Stream 队列
// 声明流队列
streamInfo, err := conn.Management().DeclareQueue(ctx, &rmq.StreamSpecification{
Name: "my-event-stream",
// 可配置最大段大小、保留策略等
})发布到 Stream
publisher, err := conn.NewPublisher(ctx, &rmq.QueueAddress{
Queue: "my-event-stream",
}, nil)
defer publisher.Close(context.Background())
// 发布多条消息到流
for i := 0; i < 100; i++ {
msg := rmq.NewMessage([]byte(fmt.Sprintf("Event #%d", i)))
msg.SetApplicationProperty("event_id", i)
msg.SetApplicationProperty("event_type", "page_view")
result, err := publisher.Publish(ctx, msg)
if err != nil {
log.Printf("发布失败: %v", err)
continue
}
// 检查结果...
_ = result
}从 Stream 消费（带偏移量控制）
// Stream 消费者支持从指定位置开始消费
consumer, err := conn.NewConsumer(ctx, "my-event-stream", &rmq.ConsumerOptions{
InitialCredits: 100, // 流可以一次拉取更多
})
// 注意：具体 Offset 设置方式取决于库版本
// 某些版本通过 StreamOptions 设置：
// consumer, err := conn.NewConsumer(ctx, "my-event-stream", &rmq.ConsumerOptions{
// StreamOptions: &rmq.StreamOptions{
// Offset: "first", // "first" | "last" | "next" | 具体数字
// },
// })Stream 过滤
// 消费时可按应用属性过滤
consumer, err := conn.NewConsumer(ctx, "my-event-stream", &rmq.ConsumerOptions{
InitialCredits: 100,
StreamOptions: &rmq.StreamOptions{
// 按过滤值消费，只接收匹配的消息
FilterValues: []string{"page_view", "click"},
// 从最早的消息开始
Offset: "first",
},
})
for {
delivery, err := consumer.Receive(ctx)
if err != nil {
break
}
msg := delivery.Message()
body := string(msg.Data()[0])
fmt.Printf("收到流消息: %s\n", body)
delivery.Accept(ctx)
}Producer 端过滤标记
// 发布时为消息打上过滤标记
msg := rmq.NewMessage([]byte("用户点击了按钮"))
msg.SetApplicationProperty("event_type", "click") // 该值可被消费者过滤
// 设置过滤器注解（RabbitMQ 将据此建立 Bloom Filter 索引）
msg.SetMessageAnnotation("x-stream-filter-value", "click")
publisher.Publish(ctx, msg)