526互联

rabbitmq Exception (504) Reason: \"channel id space exhausted\

发布时间 2023-09-12 11:27:51作者: 李若盛开
使用go的包：github.com/rabbitmq/amqp091-go
出现报错：get mq channel error    {"error": "Exception (504) Reason: \"channel id space exhausted\""}
　　ctx := context.Background()
    results, err := global.Redis.LRange(ctx, abListName, 0, -1).Result()
    if err != nil {
        global.Logger.Error("get list ABReportData error", zap.Error(err))
    }

    conn := initMqAB()
    defer conn.Close()

    for _, jsonStr := range results {
        body := []byte(jsonStr)

        // 创建Channel
        ch, err := conn.Channel()
        if err != nil {
            global.Logger.Error("get mqNew channel error", zap.Error(err))
            LogErr("get mqNew channel error: err = ", err)

            return
        }
        defer func() {
            _ = ch.Close()
        }()

        err = ch.ExchangeDeclare(exchangeNameNew, amqp.ExchangeFanout, true, false, false, false, nil)
        if err != nil {
            global.Logger.Error("MQNew ExchangeDeclare error", zap.Error(err))
            LogErr("MQNew ExchangeDeclare error: err = ", err)

            return
        }

        // 启用确认模式
        if err := ch.Confirm(false); err != nil {
            global.Logger.Error("MQNew 生产者无法启用确认模式 error", zap.Error(err))
            LogErr("MQNew 生产者无法启用确认模式 error: err = ", err)

            return
        }
        // 创建一个用于接收确认信息的通道
        confirms := ch.NotifyPublish(make(chan amqp.Confirmation, 1))

        // 发送消息
        err = ch.PublishWithContext(
            context.TODO(),
            exchangeNameNew,
            "",
            false,
            false,
            amqp.Publishing{
                ContentType: "text/plain",
                Body:        body,
            },
        )

        // 生产者等待确认信息
        confirmed := <-confirms
        if err == nil && confirmed.Ack {
            global.Logger.Info("AB---确认机制---发送MQNew成功", zap.ByteString("message", body))
        } else {
            global.Logger.Error("AB---确认机制---发送MQNew失败")
            LogErr("AB---确认机制---发送MQNew失败 error: 上报内容 = ", string(body))
        }
        time.Sleep(time.Millisecond * 200) // 休眠0.2秒
    }

原因：每个连接可创建的信道数量，同时最多支持2048个通道创建，超过就会报错。
解决思路：
将通道的创建和关闭移到循环外部，并在循环中重复使用相同的通道。这样可以避免为每个消息创建一个新通道。
在循环外创建了一个通道 ch，然后在循环中重复使用它来发送多个消息。最后，在函数退出时，我们使用 defer 关闭通道以确保资源被正确释放。这种方式可以避免创建大量不必要的通道，提高了资源利用率。
解决方案：
ctx := context.Background()
    results, err := global.Redis.LRange(ctx, abListName, 0, -1).Result()
    if err != nil {
        global.Logger.Error("get list ABReportData error", zap.Error(err))
    }

    conn := initMqAB()
    defer conn.Close()

    // 创建Channel
    ch, err := conn.Channel()
    if err != nil {
        global.Logger.Error("get mqNew channel error", zap.Error(err))
        LogErr("get mqNew channel error: err = ", err)

        return
    }
    defer func() {
        _ = ch.Close()
    }()

    err = ch.ExchangeDeclare(exchangeNameNew, amqp.ExchangeFanout, true, false, false, false, nil)
    if err != nil {
        global.Logger.Error("MQNew ExchangeDeclare error", zap.Error(err))
        LogErr("MQNew ExchangeDeclare error: err = ", err)

        return
    }

    // 启用确认模式
    if err := ch.Confirm(false); err != nil {
        global.Logger.Error("MQNew 生产者无法启用确认模式 error", zap.Error(err))
        LogErr("MQNew 生产者无法启用确认模式 error: err = ", err)

        return
    }
    // 创建一个用于接收确认信息的通道
    confirms := ch.NotifyPublish(make(chan amqp.Confirmation, 1))

    for _, jsonStr := range results {
        body := []byte(jsonStr)
        // 发送消息
        err = ch.PublishWithContext(
            context.TODO(),
            exchangeNameNew,
            "",
            false,
            false,
            amqp.Publishing{
                ContentType: "text/plain",
                Body:        body,
            },
        )

        // 生产者等待确认信息
        confirmed := <-confirms
        if err == nil && confirmed.Ack {global.Logger.Info("AB---确认机制---发送MQNew成功", zap.ByteString("message", body))
        } else {global.Logger.Error("AB---确认机制---发送MQNew失败")
            LogErr("AB---确认机制---发送MQNew失败 error: 上报内容 = ", string(body))
        }
        time.Sleep(time.Millisecond * 200) // 休眠0.2秒
    }
总结：上述操作避免了在循环中重复创建通道，从而减少了资源的浪费。但是需要注意以下几个方面以确保代码的正确性：

1、通道关闭顺序：在循环外部使用 defer 关闭通道确保资源释放，但要确保通道的关闭顺序不会导致任何问题。通常来说，关闭通道的顺序应该与它们的创建顺序相反。在上面的代码中，这个顺序是正确的，因为创建和关闭通道的代码都在循环之外。

2、并发安全：如果多个 goroutine（协程）同时使用通道，确保对通道的访问是并发安全的，可以通过使用互斥锁来实现。在上述示例中，我们只在一个 goroutine 中使用通道，因此不涉及并发问题。

3、错误处理：代码中已经包含了错误处理，它会处理通道创建、交换器声明、消息发送等可能出现的错误。确保错误处理逻辑能够适应您的应用程序需求，并记录必要的信息以便故障排除。