Golang如何抽象Channel实现细节，使Consumer接口兼容多消息中间件

要让消费者接口脱离具体MQ实现，核心是两步：抽象通用消息类型 + 封装通道转换逻辑，下面是具体实现步骤：

1. 定义通用消息类型 #

先抽取出不同MQ消息的共性字段（比如消息体、头部、ID），同时封装消息确认/拒绝的通用方法，把具体MQ的操作隐藏起来：

// Message 通用消息结构体，适配各类MQ的核心属性
type Message struct {
    Body      []byte
    Headers   map[string]interface{}
    MessageID string
    // 内部封装具体MQ的操作函数
    ackFunc  func(bool) error
    nackFunc func(bool, bool) error
}

// Ack 通用消息确认方法，支持批量确认
func (m *Message) Ack(multiple bool) error {
    if m.ackFunc != nil {
        return m.ackFunc(multiple)
    }
    return nil
}

// Nack 通用消息拒绝方法，支持批量拒绝和重新入队
func (m *Message) Nack(multiple, requeue bool) error {
    if m.nackFunc != nil {
        return m.nackFunc(multiple, requeue)
    }
    return nil
}

2. 修改通用Consumer接口 #

把原来返回的amqp.Delivery通道替换成通用Message通道，彻底和RabbitMQ解耦：

type Consumer interface {
    StartConsuming(queueName, key string) (<-chan *Message, error)
}

3. 适配RabbitMQ实现 #

在RabbitMQ的消费者实现中，先获取原生的amqp.Delivery通道，再通过goroutine把原生消息转换成通用Message，同时把RabbitMQ的Ack/Nack逻辑封装到通用消息里：

package rabbitmq

import amqp "github.com/rabbitmq/amqp091-go"

type Consumer struct {
    *Connection
}

func (c Consumer) StartConsuming(queueName, key string) (<-chan *Message, error) {
    _, err := c.Channel.QueueDeclare(
        queueName,
        true,
        false,
        false,
        false,
        nil,
    )
    if err != nil {
        return nil, err
    }

    // 省略队列绑定等业务逻辑...

    // 获取RabbitMQ原生消息通道
    amqpChan, err := c.Connection.Consume(
        queueName,
        "",
        false, // 手动确认，避免自动确认逻辑耦合
        false,
        false,
        false,
        nil,
    )
    if err != nil {
        return nil, err
    }

    // 创建通用消息通道
    msgChan := make(chan *Message)

    // 启动goroutine做类型转换，解耦原生消息与通用接口
    go func() {
        defer close(msgChan)
        for delivery := range amqpChan {
            msg := &Message{
                Body:      delivery.Body,
                Headers:   delivery.Headers,
                MessageID: delivery.MessageId,
                // 封装RabbitMQ的确认函数
                ackFunc:  delivery.Ack,
                nackFunc: delivery.Nack,
            }
            msgChan <- msg
        }
    }()

    return msgChan, nil
}

4. 切换到其他MQ的适配思路 #

如果后续要切换到MQTT等其他中间件，只需要实现Consumer接口即可：

• 获取对应MQ的原生消息通道
• 把原生消息转换成通用Message结构体
• 封装该MQ对应的消息确认/拒绝逻辑到ackFunc和nackFunc中

这样业务代码只需要依赖通用的Consumer接口和Message类型，完全不需要关心底层用的是哪种MQ，实现了真正的解耦。

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