队列与交换机
四种队列类型（Quorum/Classic/Stream/自动生成）的声明与删除、四种交换机类型（Direct/Fanout/Topic/Headers）的声明，以及队列与交换机之间的绑定操作。
队列管理
队列类型说明
| 队列类型 | Go 结构体 | 特点 |
|---|---|---|
| Quorum Queue（仲裁队列） | QuorumQueueSpecification |
高可用、基于 Raft 共识、持久化 |
| Classic Queue（经典队列） | ClassicQueueSpecification |
传统队列、兼容旧版 |
| Stream Queue（流队列） | StreamSpecification |
不可变日志、支持重放、大吞吐 |
| Auto-generated Queue（自动生成） | AutoGeneratedQueueSpecification |
服务器自动命名、临时队列 |
声明各种队列
// 获取 Management 接口
mgmt := conn.Management()
// ─── 1. 仲裁队列（推荐生产使用）───
queueInfo, err := mgmt.DeclareQueue(ctx, &rmq.QuorumQueueSpecification{
Name: "my-quorum-queue",
})
// 仲裁队列默认持久化、高可用
// ─── 2. 经典队列 ───
queueInfo, err := mgmt.DeclareQueue(ctx, &rmq.ClassicQueueSpecification{
Name: "my-classic-queue",
IsDurable: true, // 持久化，重启后保留
IsExclusive: false, // 是否独占
IsAutoDelete: false, // 消费者断开后自动删除
})
// ─── 3. 流队列 ───
queueInfo, err := mgmt.DeclareQueue(ctx, &rmq.StreamSpecification{
Name: "my-stream",
// 流队列的参数可额外配置
})
// ─── 4. 自动命名队列（临时队列）───
queueInfo, err := mgmt.DeclareQueue(ctx, &rmq.AutoGeneratedQueueSpecification{
IsExclusive: true, // 连接断开后自动删除
IsAutoDelete: true, // 消费者断开后自动删除
QueueName: "", // 留空让服务器自动命名
})
// queueInfo.Name() 可获取服务器分配的名称
fmt.Printf("自动生成的队列名: %s\n", queueInfo.Name())删除队列
// 删除指定队列，返回删除的消息数量
deletedCount, err := mgmt.DeleteQueue(ctx, "my-queue-name")
if err != nil {
log.Printf("删除队列失败: %v", err)
}
fmt.Printf("删除了 %d 条消息\n", deletedCount)清空队列
// 清空队列中的所有消息（不移除队列本身）
purgedCount, err := mgmt.PurgeQueue(ctx, "my-queue-name")
if err != nil {
log.Printf("清空队列失败: %v", err)
}
fmt.Printf("清空了 %d 条消息\n", purgedCount)交换机管理与绑定
交换机类型
| 类型 | Go 结构体 | 路由行为 |
|---|---|---|
| Direct | DirectExchangeSpecification |
精确匹配 routing key |
| Fanout | FanoutExchangeSpecification |
广播到所有绑定队列 |
| Topic | TopicExchangeSpecification |
基于模式匹配（* 和 # 通配符） |
| Headers | HeadersExchangeSpecification |
基于消息头匹配 |
声明交换机
mgmt := conn.Management()
// ─── Direct 交换机 ───
err := mgmt.DeclareExchange(ctx, &rmq.DirectExchangeSpecification{
Name: "my-direct-exchange",
IsDurable: true, // 持久化
IsAutoDelete: false, // 自动删除
})
// ─── Fanout 交换机（广播）───
err := mgmt.DeclareExchange(ctx, &rmq.FanoutExchangeSpecification{
Name: "logs-exchange",
})
// ─── Topic 交换机（模式匹配）───
err := mgmt.DeclareExchange(ctx, &rmq.TopicExchangeSpecification{
Name: "topic-exchange",
IsDurable: true,
})
// ─── Headers 交换机 ───
err := mgmt.DeclareExchange(ctx, &rmq.HeadersExchangeSpecification{
Name: "headers-exchange",
})绑定队列到交换机
// 将队列绑定到交换机
// 语法: BindQueue(ctx, 交换机名, 队列名, routingKey, 绑定参数)
err := mgmt.BindQueue(ctx, &rmq.BindingSpecification{
SourceExchange: "my-exchange", // 源交换机
DestinationQueue: "my-queue", // 目标队列
BindingKey: "info", // routing key
})
// 解绑
err := mgmt.UnbindQueue(ctx, &rmq.BindingSpecification{
SourceExchange: "my-exchange",
DestinationQueue: "my-queue",
BindingKey: "info",
})删除交换机
err := mgmt.DeleteExchange(ctx, "my-exchange-name")