Узел RabbitMQ потребляет сообщение и делает что-то для каждого сообщения - PullRequest
Новая
       45

Узел RabbitMQ потребляет сообщение и делает что-то для каждого сообщения

0 голосов
/ 04 марта 2020

Я хочу получать сообщения из службы rabbitmq, и для каждого полученного сообщения я хочу что-то делать (например: поместить это сообщение в базу данных, обработать сообщение и отправить ответ через RabbitMq через другую очередь) для каждого сообщения.

В настоящее время мой потребительский код RabbitMq выглядит следующим образом: 

const all = require('bluebird').all;
const basename = require('path').basename;


function receive() {
    const severities = process.argv.slice(2);
    if (severities.length < 1) {
        console.warn('Usage: %s [info] [warning] [error]',
            basename(process.argv[1]));
        process.exit(1);
    }
    let config = {
        protocol: 'amqp',
        hostname: 'localhost',
        port: 5672,
        username: 'rumesh',
        password: 'password',
        locale: 'en_US',
        frameMax: 0,
        heartbeat: 0,
        vhost: '/',
    };
    amqp.connect(config).then(function (conn) {
        process.once('SIGINT', function () {
            conn.close();
        });
        return conn.createChannel().then(function (ch) {
            let queue = 'test';
            let exchange = 'test-exchange';
            let key = 'python-key';
            let exchange_type = 'direct';

            let ok = ch.assertExchange(exchange, exchange_type, {durable: true});

            ok = ok.then(function () {
                return ch.assertQueue(queue, { durable: true});
            });

            ok = ok.then(function (qok) {
                const queue = qok.queue;
                return all(severities.map(function (sev) {
                    ch.bindQueue(queue, exchange, sev,{durable: true});
                })).then(function () {
                    return queue;
                });
            });

            ok = ok.then(function (queue) {
                return ch.consume(queue, logMessage, {noAck: true});
            });
            return ok.then(function () {
                console.log(' [*] Waiting for logs. To exit press CTRL+C.');
            });

            function logMessage(msg) {
                console.log(" [x] %s:'%s'",
                    msg.fields.routingKey,
                    msg.content.toString());
            }
        });
    }).catch(console.warn);
}


module.exports = receive;```

1 Ответ

2 голосов
/ 04 марта 2020

Я бы посоветовал вам создать функцию-обработчик, например, onNewMessage, которая будет вызываться каждый раз, когда вы получаете новое сообщение в очереди.

Вы можете кодировать сообщения множеством способов, поскольку вы можете отправлять двоичные данные через AMQP.

JSON - определенно один из способов отправки сообщений, с которым очень удобно иметь дело в Node.js.

Вот пример кода, который подключается к серверу, затем отправляет и получает сообщения : 

const amqp = require('amqplib');

const queue = 'test';

// Set your config here...
let config = {
    protocol: 'amqp',
    hostname: 'localhost',
    port: 5672,
    username: 'rumesh',
    password: 'password',
    locale: 'en_US',
    frameMax: 0,
    heartbeat: 0,
    vhost: '/',
};


async function start() {
    try {
        const conn = await createConnection(config);
        console.log("Connected to AMQP server.");
        let channel = await conn.createChannel();
        await channel.assertQueue(queue, { durable: true});

        startPollingForMessages(channel);
        startSendingMessages(channel);
    } catch (err) {
        console.error("start: Connection error:",err.message);
    }
}

async function createConnection(config) {
    const conn = await amqp.connect(config);

    conn.on("error", function(err) {
        console.error("Connection error:",err.message);
    });

    conn.on("close", function() {
        console.error("Connection closed:", err.message);
    });

    return conn;
}

function startSendingMessages(channel) {
    const SEND_INTERVAL = 5000;
    setInterval(() => { 
        sendMessage(channel, queue, JSON.stringify({ timestamp: new Date().toISOString(), message: " Some message" })); 
    }, SEND_INTERVAL);
}

async function sendMessage(channel, queue, messageContent) {
    console.log(`sendMessage: sending message: ${messageContent}...`);
    return channel.sendToQueue(queue, Buffer.from(messageContent))
}

function startPollingForMessages(ch) {
    ch.consume(queue, (msg) => {
        onNewMessage(msg);
        ch.ack(msg);
    });
}

function onNewMessage(msg) {
    // Do your database stuff or whatever here....
    console.log("On new message:", msg.content.toString())
}

start();
...