踩坑：mqurl=amqp://username:password@host:port/vhost，创建的vhost名称是什么就写什么，不管vhost自身有没有带/

1. amqp

go get github.com/streadway/amqp

2. 自定义结构体

  1. /**
  2.  * @Author: cyj19
  3.  * @Date: 2022/5/17 10:49
  4.  */
  6. package rabbitmq
  8. import (
  9.     "context"
  10.     "github.com/streadway/amqp"
  11. )
  13. type RabbitMQ struct {
  14.     Ctx        context.Context
  15.     Conn       *amqp.Connection // 连接
  16.     Channel    *amqp.Channel    // 管道
  17.     QueueName  string           // 队列名称
  18.     Exchange   string           // 交换机
  19.     RoutingKey string           // 路由key
  20.     MQUrl      string           // mq连接
  21. }
  23. func NewRabbitMQ(ctx context.Context, url, queueName, exchange, key string) (*RabbitMQ, error) {
  24.     var err error
  25.     r := &RabbitMQ{
  26.         Ctx:        ctx,
  27.         QueueName:  queueName,
  28.         Exchange:   exchange,
  29.         RoutingKey: key,
  30.         MQUrl:      url,
  31.     }
  32.     r.Conn, err = amqp.Dial(url)
  33.     if err != nil {
  34.         return nil, err
  35.     }
  37.     r.Channel, err = r.Conn.Channel()
  38.     if err != nil {
  39.         return nil, err
  40.     }
  42.     return r, nil
  43. }
  45. // Close 释放资源
  46. func (mq *RabbitMQ) Close() {
  47.     mq.Conn.Close()
  48.     mq.Channel.Close()
  49. }
  51. func (mq RabbitMQ) InitQueue() error {
  52.     // 声明队列
  53.     _, err := mq.Channel.QueueDeclare(
  54.         mq.QueueName,
  55.         true,  // 是否持久化
  56.         false, // 是否自动删除(前提是至少有一个消费者连接到这个队列，之后所有与这个队列连接的消费者都断开时，才会自动删除。注意：生产者客户端创建这个队列，或者没有消费者客户端与这个队列连接时，都不会自动删除这个队列)
  57.         false, // 是否为排他队列（排他的队列仅对“首次”声明的conn可见[一个conn中的其他channel也能访问该队列]，conn结束后队列删除）
  58.         false, // 是否等待服务器确认，为 true 时，无需等待服务器的确认
  59.         nil,   //额外属性
  60.     )
  61.     return err
  62. }
  64. func (mq RabbitMQ) InitExchange() error {
  65.     // 声明交换机
  66.     err := mq.Channel.ExchangeDeclare(
  67.         mq.Exchange,
  68.         "topic", //exchange type：一般用fanout、direct、topic
  69.         true,    // 是否持久化
  70.         false,   //是否自动删除（自动删除的前提是至少有一个队列或者交换器与这和交换器绑定，之后所有与这个交换器绑定的队列或者交换器都与此解绑）
  71.         false,   //设置是否内置的。true表示是内置的交换器，客户端程序无法直接发送消息到这个交换器中，只能通过交换器路由到交换器这种方式
  72.         false,   // 是否等待服务器确认，为 true 时，无需等待服务器的确认
  73.         nil,     // 额外属性
  74.     )
  76.     return err
  77. }
  79. func (mq *RabbitMQ) InitBind() error {
  80.     // 建立Binding(可建立多个绑定关系)
  81.     err := mq.Channel.QueueBind(
  82.         mq.QueueName,  // 绑定的队列名称
  83.         mq.RoutingKey, // 用于消息路由分发的key
  84.         mq.Exchange,   // 绑定的exchange名
  85.         false,         // 是否等待服务器确认，为 true 时，无需等待服务器的确认
  86.         nil,           // 额外属性
  87.     )
  88.     return err
  89. }
  91. func (mq RabbitMQ) Public(msg amqp.Publishing) error {
  92.     err := mq.Channel.Publish(mq.Exchange, mq.RoutingKey, false, false, msg)
  93.     return err
  94. }
  1. /**
  2.  * @Author: cyj19
  3.  * @Date: 2022/5/18 17:11
  4.  */
  6. package rabbitmq
  8. import (
  9.     "github.com/streadway/amqp"
  10.     "time"
  11. )
  13. type PublicMessage struct {
  14.     StartTime time.Time
  15.     Msg       amqp.Publishing
  16. }

3. 生产者

  1. /**
  2.  * @Author: cyj19
  3.  * @Date: 2022/5/17 11:35
  4.  */
  6. package main
  8. import (
  9.     "context"
  10.     "day20220517-02/pkg/rabbitmq"
  11.     "github.com/rs/xid"
  12.     "github.com/streadway/amqp"
  13.     "log"
  14.     "time"
  15. )
  17. func main() {
  18.     // 声明生成者
  19.     url := "amqp://admin:admin@localhost:5672//go-mq"
  20.     mq, err := rabbitmq.NewRabbitMQ(context.Background(), url, "go-queue", "go-exchange", "go-key")
  21.     if err != nil {
  22.         log.Fatalln(err)
  23.     }
  24.     defer mq.Close()
  26.     err = mq.InitQueue()
  27.     if err != nil {
  28.         log.Fatalln(err)
  29.     }
  31.     err = mq.InitExchange()
  32.     if err != nil {
  33.         log.Fatalln(err)
  34.     }
  36.     err = mq.InitBind()
  37.     if err != nil {
  38.         log.Fatalln(err)
  39.     }
  41.     err = mq.Channel.Confirm(false)
  42.     if err != nil {
  43.         log.Fatalln(err)
  44.     }
  45.     confirm := mq.Channel.NotifyPublish(make(chan amqp.Confirmation))
  47.     ticker := time.NewTicker(20 * time.Second)
  48.     defer func() {
  49.         ticker.Stop()
  50.     }()
  51.     deliveryMap := make(map[uint64]*rabbitmq.PublicMessage)
  53.     var deliveryTag uint64 = 1
  54.     var pMsg *rabbitmq.PublicMessage
  56.     go func() {
  57.         for i := 0; i < 10; i++ {
  58.             pMsg = &rabbitmq.PublicMessage{
  59.                 StartTime: time.Now(),
  60.                 Msg: amqp.Publishing{ // 发送的消息，固定有消息体和一些额外的消息头，包中提供了封装对象
  61.                     MessageId:   xid.New().String(),    // 消息ID
  62.                     ContentType: "text/plain",          // 消息内容的类型
  63.                     Body:        []byte("hello cyj19"), // 消息内容
  64.                 },
  65.             }
  66.             // 4.发送消息
  67.             errv := mq.Public(pMsg.Msg)
  68.             if errv != nil {
  69.                 return
  70.             }
  71.             deliveryMap[deliveryTag] = pMsg
  72.             deliveryTag++
  73.         }
  75.     }()
  77.     for {
  78.         select {
  79.         case <-mq.Ctx.Done():
  80.             return
  81.         case c, ok := <-confirm:
  82.             if !ok {
  83.                 log.Println("[RABBITMQ_CLIENT]", "client Publish notify channel error")
  84.                 return
  85.             }
  86.             log.Println("RabbitMQ ack")
  87.             pMsg = deliveryMap[c.DeliveryTag]
  88.             // fmt.Println("DeliveryTag:", c.DeliveryTag)
  89.             delete(deliveryMap, c.DeliveryTag)
  91.         case <-ticker.C:
  92.             now := time.Now()
  93.             // 遍历消息表，
  94.             for key := range deliveryMap {
  96.                 pMsg = deliveryMap[key]
  97.                 if pMsg != nil {
  98.                     // 消息超时还没收到ack，重发一次消息
  99.                     if now.Sub(pMsg.StartTime.Add(10*time.Second)) > 0 {
  100.                         delete(deliveryMap, key)
  101.                         mq.Public(pMsg.Msg)
  102.                     }
  103.                 }
  105.             }
  107.         }
  108.     }
  109. }

4. 消费者

  1. package main
  3. func main(){
  4.     // 1. 初始化MQ
  5.     ctx := context.Background()
  6.     url := "amqp://username:password@127.0.0.1:5672/hello-mq"
  7.     mq, err := rabbit.NewRabbitMQ(ctx, url, "hello-queue", "hello-exchange", "hello-key")
  8.     if err != nil {
  9.         log.Fatalln(err)
  10.     }
  11.     defer mq.Close()
  13.     err = mq.InitQueue()
  14.     if err != nil {
  15.         log.Fatalln(err)
  16.     }
  18.     // 3.从队列获取消息（消费者只关注队列）consume方式会不断的从队列中获取消息
  19.     msgChan, err := mq.Channel.Consume(
  20.         mq.QueueName, // 队列名
  21.         "",           // 消费者名，用来区分多个消费者，以实现公平分发或均等分发策略
  22.         false,        // 是否自动应答
  23.         false,        // 是否排他
  24.         false,        // 是否接收只同一个连接中的消息，若为true，则只能接收一个conn中发送的消息
  25.         false,         // 队列消费是否阻塞
  26.         nil,          // 额外属性
  27.     )
  28.     if err != nil {
  29.         log.Println("获取消息失败", err)
  30.         return
  31.     }
  33.     // 优雅退出
  34.     quit := make(chan os.Signal, 1)
  35.     signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
  37.     for msg := range msgChan {
  39.         // 消费者对重复消息进行过滤，使用redis存储消息
  40.         ok, err := rdb.Exists(ctx, msg.MessageId).Result()
  41.         if err != nil {
  42.             log.Println(err)
  43.             msg.Ack(false) // 主动应答
  44.             continue
  45.         }
  46.         if ok == 1 {
  47.             log.Printf("message:%s does consumed", msg.MessageId)
  48.             msg.Ack(false) // 主动应答
  49.             continue
  50.         }
  51.         // To DO ...
  52.         log.Println(string(msg.Body))
  53.         // 加入到redis
  54.         rdb.Set(ctx, msg.MessageId, "", 10*time.Minute)
  55.         msg.Ack(false) // 主动应答
  57.     }
  59.     <-quit
  60. }