RabbitMQ 的实现流程

安装

import "github.com/go-redis/redis/v7"
go mod tidy

创建交换器 - Add exchange

// ExchangeDelete removes the named exchange from the server.
func (r *RabbitMQ) DeleteExchange(name string) (err error) {
    err = r.channel.ExchangeDelete(name, false, false)
    if err != nil {
        log.Printf("[amqp] delete exchange error: %s\n", err)
        return err
    }
    return nil
}

创建队列 - Add a new queue

// QueueDeclare declares a queue to hold messages and deliver to consumers.
func (r *RabbitMQ) DeclareQueue(name string, durable, autodelete, exclusive, nowait bool) (err error) {
    _, err = r.channel.QueueDeclare(name, durable, autodelete, exclusive, nowait, nil)
    if err != nil {
        log.Printf("[amqp] declare queue error: %s\n", err)
        return err
    }
    return nil
}

路由绑定 - Bindings

// QueueBind binds an exchange to a queue
func (r *RabbitMQ) BindQueue(queue, exchange string, keys []string, nowait bool) (err error) {
    for _, key := range keys {
        if err = r.channel.QueueBind(queue, key, exchange, nowait, nil); err != nil {
            log.Printf("[amqp] bind queue error: %s\n", err)
            return err
        }
    }
    return nil
}

发布消息 - Publish message

func (m *mqController) PublishHandler(w http.ResponseWriter, r *http.Request) {
    //...
    err = RabbitMQ.Publish(MessageEntity.Exchange, MessageEntity.Key, MessageEntity.DeliveryMode, MessageEntity.Priority, MessageEntity.Body);
    //...   
}

消费消息 - Consume message

func (m *mqController) QueueHandler(w http.ResponseWriter, r *http.Request) {
    //...
    err := RabbitMQ.ConsumeQueue(name, message);
    //...
}

总结

• 使用RabbitMQ消息可靠性投递
• 示例代码: golang-gin-api: rabbitmq