运维与最佳实践

// ✓ 正确：使用单个 Environment 实例
env := rmq.NewEnvironment(uri, nil)
// 整个应用共享一个 Environment

// ✓ 正确：适当使用连接
conn1, _ := env.NewConnection(ctx)  // 发布用
conn2, _ := env.NewConnection(ctx)  // 消费用

// ✗ 错误：每个操作都创建新的 Environment
for i := 0; i < 100; i++ {
    env := rmq.NewEnvironment(uri, nil)  // 浪费资源
    // ...
}

// ✓ 正确处理发布结果
result, err := publisher.Publish(ctx, msg)
if err != nil {
    // 网络错误等
    log.Printf("发布失败: %v", err)
    return
}
switch result.Outcome.(type) {
case *rmq.StateAccepted:
    // 成功
case *rmq.StateRejected:
    // 消息被拒绝，可能需要重试或记录
    log.Println("消息被拒绝")
default:
    log.Printf("意外状态: %v", result.Outcome)
}

// ✓ 正确处理消费错误
delivery, err := consumer.Receive(ctx)
if err != nil {
    if errors.Is(err, context.Canceled) {
        // 正常退出
        return
    }
    // 连接错误，触发重连
    reconnect()
}

// 清理顺序：从内到外
// Publisher → Consumer → Connection → Environment
defer publisher.Close(ctx)
defer consumer.Close(ctx)
defer conn.Close(ctx)
defer env.CloseConnections(ctx)

// 1. 批量发布时复用 Publisher
publisher, _ := conn.NewPublisher(ctx, addr, nil)
for _, msg := range messages {
    publisher.Publish(ctx, msg) // 复用同一个 publisher
}

// 2. 合理设置 Credits 以提升消费吞吐
// 处理快的消费者加大 Credits
consumer, _ := conn.NewConsumer(ctx, "queue", &rmq.ConsumerOptions{
    InitialCredits: 100,  // 适合快速处理的消费者
})

// 3. 使用多个 Consumer 实现并行消费
for i := 0; i < 5; i++ {
    go func(id int) {
        consumer, _ := conn.NewConsumer(ctx, "queue", &rmq.ConsumerOptions{
            InitialCredits: 5,
        })
        for { /* 消费循环 */ }
    }(i)
}

// 定期检查连接状态
type HealthChecker struct {
    env  *rmq.Environment
    conn *rmq.Connection
    mu   sync.RWMutex
}

func (hc *HealthChecker) IsHealthy(ctx context.Context) bool {
    hc.mu.RLock()
    defer hc.mu.RUnlock()

    if hc.conn == nil {
        return false
    }

    // 通过尝试声明一个不存在的队列来验证连接
    // 如果连接正常，会收到队列不存在的错误（而非连接错误）
    _, err := hc.conn.Management().DeclareQueue(ctx,
        &rmq.QuorumQueueSpecification{Name: "health-check-probe"})
    // 注意：这种方式会实际创建队列，生产环境建议用其他方式
    return err == nil
}

// 启动健康检查 goroutine
func (hc *HealthChecker) Start(ctx context.Context, interval time.Duration) {
    go func() {
        ticker := time.NewTicker(interval)
        defer ticker.Stop()

        for {
            select {
            case <-ctx.Done():
                return
            case <-ticker.C:
                if !hc.IsHealthy(ctx) {
                    log.Println("⚠️ RabbitMQ 连接不健康，触发重连...")
                    // 触发重连逻辑
                }
            }
        }
    }()
}

func gracefulShutdown() {
    // 1. 创建可取消的 context
    ctx, cancel := context.WithCancel(context.Background())

    // 2. 注册信号处理
    sig := make(chan os.Signal, 1)
    signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM)

    // 3. 正常运行业务...
    go runConsumer(ctx, consumer)

    // 4. 等待退出信号
    s := <-sig
    log.Printf("收到信号 %v，开始优雅关闭...", s)

    // 5. 取消 context，通知所有 goroutine 停止
    cancel()

    // 6. 等待正在处理的消息完成（设置超时）
    shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 30*time.Second)
    defer shutdownCancel()

    // 7. 按顺序关闭资源：Consumer → Publisher → Connection → Environment
    log.Println("关闭 Consumer...")
    if err := consumer.Close(shutdownCtx); err != nil {
        log.Printf("关闭 Consumer 失败: %v", err)
    }

    log.Println("关闭 Publisher...")
    if err := publisher.Close(shutdownCtx); err != nil {
        log.Printf("关闭 Publisher 失败: %v", err)
    }

    log.Println("关闭 Connection...")
    if err := conn.Close(shutdownCtx); err != nil {
        log.Printf("关闭 Connection 失败: %v", err)
    }

    log.Println("关闭 Environment...")
    if err := env.CloseConnections(shutdownCtx); err != nil {
        log.Printf("关闭 Environment 失败: %v", err)
    }

    log.Println("✅ 优雅关闭完成")
}

// 监控消费延迟——消息从发布到被消费的时间差
func monitorLag(ctx context.Context, consumer *rmq.Consumer) {
    go func() {
        ticker := time.NewTicker(10 * time.Second)
        defer ticker.Stop()

        for {
            select {
            case <-ctx.Done():
                return
            case <-ticker.C:
                // 记录上一次成功消费的时间
                // 如果超过阈值（如 30 秒）没有消费到消息，触发告警
                // 注意：kafka-go 的 Reader 有 Lag 方法，但 AMQP 1.0 Go 客户端
                // 需要通过 Management API 查询队列状态来获取消息积压数
            }
        }
    }()
}

// 1. 开启详细日志
// 在 Environment 创建时注入自定义 Logger（如果库支持）

// 2. 检查消息生命周期
msg := rmq.NewMessage([]byte("test"))
msg.SetMessageID("debug-msg-001")
msg.SetCreationTime(time.Now())
// 通过 MessageID 追踪消息在 Broker 和消费者之间的流转

// 3. 使用 RabbitMQ Management UI
// 默认地址: http://localhost:15672
// 可查看队列状态、消息速率、连接信息等

// 4. 消费端记录详细的结算信息
delivery, _ := consumer.Receive(ctx)
log.Printf("收到消息: ID=%s, DeliveryCount=%d, State=%v",
    delivery.Message().MessageID(),
    delivery.Message().DeliveryCount(),
    delivery.DeliveryState())