nodejs amqplib 使用

生产者

class RabbitMq {
  constructor(options) {
    this.ex = 'XXX'
    this.exType = 'direct'
    this.durable = true
    this.routeKey = 'XX'
    this.autoDelete = true
    this.q = 'hello'
  }

  async send() {
    const conn = await amqp.connect(url)
    const msg = JSON.stringify({ a: "aa" })

    try {
      // const ch = await conn.createChannel()
      // 确认消息发送 ok 猜测是开启 confirm 机制,对应的监听函数是什么呢?
      const ch = await conn.createConfirmChannel()
      const res = await ch.assertExchange(this.ex, this.exType, { durable: this.durable })

      var flag = 0
      while(flag < 4) {
// 实现消息持久化, 要exchange,queue,msg 三者同时持久化
/*
如果exchange根据自身类型和消息routeKey无法找到一个符合条件的queue,
那么会调用basic.return方法将消息返回给生产者(Basic.Return + Content-Header + Content-Body);
当mandatory设置为false时,出现上述情形broker会直接将消息扔掉
*/
        ch.publish(this.ex, this.routeKey, Buffer.from(msg), {
          persistent: true, // 消息持久化
          mandatory: true
        })
  // 确认消息已经入队, 返回错误 是啥样? 错误怎么处理?直接close?
        const res2 = await ch.waitForConfirms()
        console.log('==res2==', res2)

        console.log(" [x] Sent '%s'", msg);

        await timeout(2000)
        flag++
      }
      ch.close()
    } catch (e) {
      console.log('==e==', e)
      ch.close()
    }
  }
}

const rabbit = new RabbitMq({})

rabbit.send()

消费端

class RabbitMq {
  constructor(options) {
    this.ex = 'XXX'
    this.exType = 'direct'
    this.durable = true
    this.routeKey = 'XX'
    this.autoDelete = true
    this.q = 'hello'
  }

  async send() {
    const conn = await amqp.connect(url)

    try {
      const ch = await conn.createChannel()
      // 确认消息发送 ok
      const res = await ch.assertExchange(this.ex, this.exType, { durable: this.durable })
      // 此处 q 置空,用的是rabbitmq自动生成的队列名, exclusive 是生成排他队列, 连接断开后就会自动删除
      const q = await ch.assertQueue('', { exclusive: false })

      console.log('==q=', q)
      // 队列绑定 exchange
      ch.bindQueue(q.queue, this.ex, this.routeKey)

      ch.consume(q.queue, msg => {
        console.log('收到消息: ', msg)
         // 发送确认消息
        ch.ack(msg)
      }, { noAck: false })

      // ch.close()
    } catch (e) {
      console.log('==e==', e)
      ch.close()
    }
  }
}

const rabbit = new RabbitMq({})

rabbit.send()
©著作权归作者所有,转载或内容合作请联系作者
平台声明:文章内容(如有图片或视频亦包括在内)由作者上传并发布,文章内容仅代表作者本人观点,简书系信息发布平台,仅提供信息存储服务。

推荐阅读更多精彩内容

  • Spring Cloud为开发人员提供了快速构建分布式系统中一些常见模式的工具(例如配置管理,服务发现,断路器,智...
    卡卡罗2017阅读 134,948评论 18 139
  • 嘿,老朋友,又见面了! 算起来这是我们第六次相见,是第五次我们共同度过一段难忘的时光,其中有一次是我乘飞机飞越滇藏...
    水竹清月阅读 401评论 0 0
  • 1.今天女友终于休假,我们俩不约而同的准备一起去看《金刚狼3》。剧情虽然让中国电影局删减的狗血至极,但也把我们感动...
    卢不斯拍个人短片阅读 606评论 3 5
  • 【从本视频中学到最重要的概念】 Try to understand the structure of the ar...
    晓猪佩琪阅读 190评论 3 0