RabbitMQ RPC Go客户端仅首次生效，二次调用超时求助

问题 #

在Go服务中参照RabbitMQ官方教程实现RPC客户端，调用示例如下：

func DoSomethingWithRpc() {
 ...
 rpcResponse, err := rmq_helpers.SendRpcMessage(
    ts.publisher, // 存储连接信息的结构体
    data,
 )
 // 处理rpcResponse
 ...
}

对应的SendRpcMessage函数实现如下。问题是客户端仅首次调用正常，第二次调用DoSomethingWithRpc()时会触发超时（无超时逻辑则阻塞线程）。但第二次调用时，RPC服务端能收到事件，CorrelationID和ReplyTo队列值均正确，RabbitMQ控制台也能看到回复消息已在队列中。

func SendRpcMessage(
    client *rmq.Client,
    body []byte,
) (res *amqp.Delivery, err error) {
    ch := client.GetChannel()

    if ch == nil {
        err = fmt.Errorf("Channel is nil")
        return
    }

    replyQueue, err := client.GetChannel().QueueDeclare(
        "",
        false,
        false,
        true,
        false,
        nil,
    )

    if err != nil {
        fmt.Println("Error declaring queue: ", err)
        return nil, err
    }

    msgs, err := client.GetChannel().Consume(
        replyQueue.Name,
        "",
        true,
        false,
        false,
        false,
        nil,
    )

    if err != nil {
        fmt.Println("Error consuming messages: ", err)
        return nil, err
    }

    corrId := fmt.Sprintf("%d", time.Now().UnixNano())
    fmt.Println("Correlation ID: ", corrId, "ReplyTo: ", replyQueue.Name)

    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    defer cancel()

    err = client.GetChannel().PublishWithContext(ctx,
        "",
        client.GetQueueName(),
        false,
        false,
        amqp.Publishing{
            ContentType:   "application/json",
            Body:          body,
            ReplyTo:       replyQueue.Name,
            CorrelationId: corrId,
        },
    )

    if err != nil {
        return nil, err
    }

    timeout := time.After(5 * time.Second)

    for {
        select {
        case d := <-msgs:
            if d.CorrelationId == corrId {
                // client.GetChannel().QueueDelete(replyQueue.Name, false, false, false)
                res = &d
                return  
            }
        case <-timeout:
            err = fmt.Errorf("Timeout")
            return
        }

    }
}

问题分析与修复 #

核心错误原因 #

• 旧消费器未清理，抢占回复消息：每次调用SendRpcMessage时，都会在同一个通道上创建新的消费者，但前一次调用的消费者并未被取消。RabbitMQ会将回复队列的消息随机分发给通道上的所有活跃消费者，导致第二次请求的回复被第一次调用遗留的消费器接收，当前请求的msgs通道永远收不到消息，最终超时。
• 匿名队列的自动删除未生效：虽然声明队列时设置了autoDelete: true，但由于旧消费器未关闭，队列始终存在活跃消费者，无法被自动删除，进一步加剧了消息分发的混乱。

修复方案 #

1. 每次调用后主动取消当前消费者：给每个消费者设置唯一标签，函数退出时调用Cancel方法清理消费器，避免通道上残留无效的消费实例。
2. 确保消费资源被正确释放：通过defer语句保证无论函数正常返回还是报错，都能取消当前消费者，让临时回复队列在无活跃消费者时被自动删除。

修复后的SendRpcMessage函数：

func SendRpcMessage(
    client *rmq.Client,
    body []byte,
) (res *amqp.Delivery, err error) {
    ch := client.GetChannel()
    if ch == nil {
        err = fmt.Errorf("Channel is nil")
        return
    }

    // 声明临时回复队列
    replyQueue, err := ch.QueueDeclare(
        "",    // 匿名队列
        false, // durable
        false, // delete when unused
        true,  // exclusive
        false, // noWait
        nil,
    )
    if err != nil {
        fmt.Println("Error declaring queue: ", err)
        return nil, err
    }

    // 生成唯一消费者标签，用于后续取消消费
    corrId := fmt.Sprintf("%d", time.Now().UnixNano())
    consumerTag := fmt.Sprintf("rpc-consumer-%s", corrId)
    
    msgs, err := ch.Consume(
        replyQueue.Name,
        consumerTag, // 非空消费者标签
        true,        // autoAck
        false,       // exclusive
        false,       // noLocal
        false,       // noWait
        nil,
    )
    if err != nil {
        fmt.Println("Error consuming messages: ", err)
        return nil, err
    }

    // 延迟取消消费者，确保函数退出时清理资源
    defer func() {
        if cancelErr := ch.Cancel(consumerTag, false); cancelErr != nil {
            fmt.Println("Error canceling consumer: ", cancelErr)
        }
    }()

    fmt.Println("Correlation ID: ", corrId, "ReplyTo: ", replyQueue.Name)

    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    defer cancel()

    // 发布RPC请求
    err = ch.PublishWithContext(ctx,
        "",
        client.GetQueueName(),
        false,
        false,
        amqp.Publishing{
            ContentType:   "application/json",
            Body:          body,
            ReplyTo:       replyQueue.Name,
            CorrelationId: corrId,
        },
    )
    if err != nil {
        return nil, err
    }

    timeout := time.After(5 * time.Second)

    for {
        select {
        case d := <-msgs:
            if d.CorrelationId == corrId {
                res = &d
                return
            }
        case <-timeout:
            err = fmt.Errorf("Timeout")
            return
        }
    }
}

额外优化建议 #

• 通道线程安全：如果多个goroutine同时调用该函数，复用同一个通道会存在线程安全问题（amqp通道并非线程安全），建议为每个请求创建独立通道，或使用互斥锁保护通道操作。
• 复用回复队列：若同一客户端发起多次RPC请求，可以复用一个固定的回复队列，减少队列创建开销，但需严格通过CorrelationId过滤消息，避免不同请求的回复混淆。

内容的提问来源于stack exchange，提问作者Dilshan