发布消息
Publisher 的创建（QueueAddress / ExchangeAddress 两种目标地址）、发送消息与结果检查、消息属性完整列表，以及应用属性（Application Properties）与消息注解（Message Annotations）。
发布消息
创建 Publisher
// ─── 发布到指定队列 ───
publisher, err := conn.NewPublisher(ctx, &rmq.QueueAddress{
Queue: "my-queue",
}, nil)
defer publisher.Close(context.Background())
// ─── 发布到交换机 ───
publisher, err := conn.NewPublisher(ctx, &rmq.ExchangeAddress{
Exchange: "my-exchange",
Key: "routing-key", // 路由键
}, nil)
defer publisher.Close(context.Background())地址类型对比
| 地址类型 | 结构体 | 使用场景 |
|---|---|---|
| QueueAddress | &rmq.QueueAddress{Queue: "name"} |
直接发布到队列 |
| ExchangeAddress | &rmq.ExchangeAddress{Exchange: "name", Key: "rk"} |
通过交换机路由 |
发送消息
// ─── 基础发送 ───
message := rmq.NewMessage([]byte("Hello, RabbitMQ!"))
result, err := publisher.Publish(ctx, message)
if err != nil {
log.Fatalf("发布失败: %v", err)
}
// ─── 检查发送结果 ───
switch result.Outcome.(type) {
case *rmq.StateAccepted:
fmt.Println("✓ 消息已被 Broker 接受")
case *rmq.StateReleased:
fmt.Println("✗ 消息被 Broker 释放（未确认）")
case *rmq.StateRejected:
fmt.Println("✗ 消息被 Broker 拒绝")
default:
fmt.Printf("未知结果: %v\n", result.Outcome)
}带属性的消息
msg := rmq.NewMessage([]byte("重要的消息"))
// 设置消息属性
msg.SetContentType("application/json")
msg.SetContentEncoding("utf-8")
msg.SetCorrelationID("req-12345")
msg.SetMessageID("msg-67890")
msg.SetReplyTo("response-queue")
msg.SetSubject("order.created")
msg.SetPriority(5) // 优先级 0-9
msg.SetTTL(60000) // 生存时间（毫秒）
msg.SetCreationTime(time.Now())
// 设置应用自定义属性
msg.SetApplicationProperty("source", "web-app")
msg.SetApplicationProperty("version", "1.0.0")
msg.SetApplicationProperty("userId", 12345)
// 设置消息注解
msg.SetMessageAnnotation("x-opt-enqueue-time", time.Now().UnixMilli())
// 发布
result, err := publisher.Publish(ctx, msg)发布到不同目标（单 Publisher 多目标）
// 同一个 publisher，通过消息的 To 属性指定目标
publisher, err := conn.NewPublisher(ctx, nil, nil) // address 可以为 nil
msg1 := rmq.NewMessage([]byte("发给队列A"))
msg1.SetTo("queue-a") // 指定目标队列
msg2 := rmq.NewMessage([]byte("发给队列B"))
msg2.SetTo("queue-b")
result1, err := publisher.Publish(ctx, msg1)
result2, err := publisher.Publish(ctx, msg2)消息属性与高级选项
消息属性完整列表
| 方法 | 类型 | 说明 |
|---|---|---|
SetContentType(ct) |
string | 内容类型，如 "application/json" |
SetContentEncoding(ce) |
string | 内容编码，如 "utf-8" |
SetMessageID(id) |
string | 消息唯一 ID |
SetCorrelationID(cid) |
string | 关联 ID（用于 RPC） |
SetReplyTo(addr) |
string | 回复目标地址 |
SetSubject(subj) |
string | 消息主题 |
SetPriority(p) |
int | 优先级（0-9，9 最高） |
SetTTL(ms) |
int | 生存时间（毫秒） |
SetCreationTime(t) |
time.Time | 创建时间 |
SetGroupID(gid) |
string | 消息分组 ID |
SetGroupSequence(seq) |
int | 消息分组序列号 |
SetReplyToGroupID(rgid) |
string | 回复分组 ID |
SetTo(addr) |
string | 目标地址 |
SetAbsoluteExpiryTime(t) |
time.Time | 绝对过期时间 |
SetUserID(uid) |
string | 用户 ID |
应用属性与注解
msg := rmq.NewMessage([]byte("body"))
// ─── 应用属性（Application Properties）───
// 键值对，可用于业务逻辑
msg.SetApplicationProperty("event_type", "order.created")
msg.SetApplicationProperty("order_id", "ORD-12345")
msg.SetApplicationProperty("amount", 99.99)
msg.SetApplicationProperty("retry_count", 0)
// 读取应用属性
if eventType, ok := msg.ApplicationProperty("event_type"); ok {
fmt.Printf("事件类型: %s\n", eventType)
}
// ─── 消息注解（Message Annotations）───
// 用于 RabbitMQ 特定的功能，如消息延迟
msg.SetMessageAnnotation("x-delay", 5000) // 延迟 5 秒投递
msg.SetMessageAnnotation("x-opt-enqueue-time", time.Now().UnixMilli())