如何使用 amqp 库重用 "setup structure" 之外的 RabbitMQ 连接和通道？

Question

我正在尝试使用 amqp 库构建一个简单的 node.js 客户端，它打开一个连接，然后打开一个到 RabbitMQ 服务器的通道。我想重复使用相同的连接和通道来发送多条消息。主要问题是，我不想在 ceateChannel() 函数的回调函数中编写我的整个代码。

如何在回调函数之外重用通道并确保在使用通道之前回调函数已完成？

回调方式和promise方式我都试过了，但是我不能让他们中的任何一个工作。在使用回调方法时我运行陷入了描述的问题。

当使用 promises 时，我遇到的问题是我无法在 .then() 函数之外保留连接和通道的引用，因为传递的变量在设置连接和通道后会被销毁。


amqp.connect('amqp://localhost', (err, conn) => {

  if (err !== null) return console.warn(err);
  console.log('Created connection!');

  conn.createChannel((err, ch) => {

    if (err !== null) return console.warn(err);
    console.log('Created channel!');

    //this is where I would need to write the code that uses the variable "ch"
    //but I want to move the code outside of this structure, while making sure
    //this callback completes before I try using "ch"

  });
});


    amqp.connect('amqp://localhost').then((conn) => {
      return conn.createChannel();
    }).then((ch) => {
      this.channel = ch;
      return ch.assertQueue('', {}).then((ok) => {
        return this.queueName = ok.queue;  
      });
    }).catch(console.warn);

Answer 1

为什么不使用 async\await？

const conn = await amqp.connect('amqp://localhost');
const ch = await conn.createChannel();
// after that you can use ch anywhere, don't forget to handle exceptions

此外，如果您使用 amqplib，请不要忘记处理 close 和内部 error 事件，例如：

conn.on('error', function (err) {
    console.log('AMQP:Error:', err);
});
conn.on('close', () => {
    console.log("AMQP:Closed");
});

Answer 2

试试 class，像这样：

RabbitConnection.js

const amqp = require('amqplib');

const RabbitSettings = {
    protocol: 'amqp',
    hostname: 'localhost',
    port: 5672,
    username: 'guest',
    password: 'guest',
    authMechanism: 'AMQPLAIN',
    vhost: '/',
    queue: 'test'
}

class RabbitConnection {

    constructor() {
        RabbitConnection.createConnection();
        this.connection = null;
        this.channel = null;
    }

    static getInstance() {
        if (!RabbitConnection.instance) {
            RabbitConnection.instance = new RabbitConnection();
        }
        return RabbitConnection.instance;
    }
    //create connection to rabbitmq
    static async createConnection() {
        try {
            this.connection = await amqp.connect(`${RabbitSettings.protocol}://${RabbitSettings.username}:${RabbitSettings.password}@${RabbitSettings.hostname}:${RabbitSettings.port}${RabbitSettings.vhost}`);
            this.channel = await this.connection.createChannel();
            this.channel.assertQueue(RabbitSettings.queue);
            console.log('Connection to RabbitMQ established');
        } catch (error) {
            console.log(error);
        }
    }
    //send message to rabbitmq queue
    static async sendMessage(message, queueName) {
        try {
            let msg = await this.channel.sendToQueue(queueName, Buffer.from(message));
            console.log('Message sent to RabbitMQ');
            return msg;
        } catch (error) {
            console.log(error);
        }
    }
}

module.exports = { RabbitConnection };

ServerExpress.js

const express = require('express');
const { RabbitConnection } = require('./RabbitConnection');

const serverUp = () => {
    const app = express();
    app.get('/', (req, res) => {
        RabbitConnection.sendMessage('Hello World', 'test');
        res.send('Hello World!');
    });
    app.listen(3000, () => {
        console.log('Server is running on port 3000');
    });
};

module.exports = { serverUp };

index.js

const { RabbitConnection } = require("./RabbitConnection");
const { serverUp } = require("./ServerExpress");

serverUp();
RabbitConnection.getInstance();

如何使用 amqp 库重用 "setup structure" 之外的 RabbitMQ 连接和通道？

How do I reuse a RabbitMQ connection and channel outside of the "setup structure" with the amqp library?

javascript

amqp

rabbitmq

node.js

promise