消费与消息确认

中间件

RabbitMQ

消费与消息确认

Consumer 的创建与接收循环、基于 Credits（信用额度）的流量控制，以及 AMQP 1.0 的 Settlement（结算）机制——Accept / Discard / Requeue 三种结算方式与结算状态详解。

消费消息

创建 Consumer

// ─── 基础消费者 ───
consumer, err := conn.NewConsumer(ctx, "my-queue", nil)
defer consumer.Close(context.Background())

// ─── 带流量控制的消费者 ───
consumer, err := conn.NewConsumer(ctx, "task-queue", &rmq.ConsumerOptions{
    InitialCredits: 10,  // 初始信用额度（类似 prefetch）
    // InitialCredits 控制同时最多有多少条未确认消息
})

接收消息循环

for {
    delivery, err := consumer.Receive(ctx)
    if err != nil {
        if errors.Is(err, context.Canceled) {
            fmt.Println("消费者正常退出")
            return
        }
        log.Printf("接收消息失败: %v", err)
        continue
    }

    // 获取消息内容
    msg := delivery.Message()

    // Data 是 [][]byte 类型，通常取第一个元素
    body := string(msg.Data()[0])
    fmt.Printf("收到消息: %s\n", body)

    // 读取消息属性
    fmt.Printf("  ContentType: %s\n", msg.ContentType())
    fmt.Printf("  MessageID: %s\n", msg.MessageID())
    fmt.Printf("  CorrelationID: %s\n", msg.CorrelationID())
    fmt.Printf("  Subject: %s\n", msg.Subject())

    // 读取应用属性
    if val, ok := msg.ApplicationProperty("source"); ok {
        fmt.Printf("  应用属性 source: %v\n", val)
    }

    // 确认消息（必须！）
    err = delivery.Accept(ctx)
    if err != nil {
        log.Printf("确认消息失败: %v", err)
    }
}

消费者的流量控制（Credits）

// AMQP 1.0 使用 Credits（信用额度）替代 AMQP 0.9.1 的 Prefetch

// 初始化时设置初始信用额度
consumer, err := conn.NewConsumer(ctx, "queue", &rmq.ConsumerOptions{
    InitialCredits: 5,  // 最多 5 条未确认消息
})

// 运行时动态授予信用额度
// 每调用一次 Receive 并 Accept 后，自动恢复 1 个 credit
// 也可以显式授予更多 credits
err = consumer.GrantCredits(ctx, 10)  // 额外授予 10 个 credits

消息确认机制（Settlement）

三种结算方式

AMQP 1.0 使用 Settlement（结算） 而非传统的 Ack/Nack：

方法	含义	类比 AMQP 0.9.1
`delivery.Accept(ctx)`	确认消息处理成功，从队列移除	`Ack()`
`delivery.Discard(ctx)`	丢弃消息（不重新入队）	`Nack(requeue=false)`
`delivery.Requeue(ctx)`	拒绝消息并重新入队	`Nack(requeue=true)`

使用示例

delivery, err := consumer.Receive(ctx)

// 处理消息
msg := delivery.Message()
body := string(msg.Data()[0])

if processMessage(body) {
    // 成功处理 → 确认
    err = delivery.Accept(ctx)
} else if isInvalidMessage(body) {
    // 无效消息 → 丢弃（不入队）
    err = delivery.Discard(ctx)
} else {
    // 暂时无法处理 → 重新入队等待重试
    err = delivery.Requeue(ctx)
}

结算状态详解

// 检查消息的结算状态
delivery, _ := consumer.Receive(ctx)

// 消息的 DeliveryState 表示当前结算状态
state := delivery.DeliveryState()

switch s := state.(type) {
case *rmq.StateAccepted:
    fmt.Println("消息已被接受")
case *rmq.StateRejected:
    fmt.Println("消息已被拒绝")
case *rmq.StateReleased:
    fmt.Println("消息已被释放")
case *rmq.StateModified:
    fmt.Printf("消息被修改: deliveryFailed=%v, undeliverableHere=%v\n",
        s.DeliveryFailed, s.UndeliverableHere)
}

发布消息消息收发模式