Analysis of how NodeJS operates message queue RabbitMQ

This article mainly introduces the analysis of how NodeJS operates the message queue RabbitMQ. It has certain reference value. Now it is shared with everyone. Friends in need can refer to it

1 . What is a message queue? Message refers to data transmitted between applications. Messages can be very simple, containing only text strings, or more complex, possibly containing embedded objects.

Message Queue (Message Queue) is a communication method between applications. Messages can be returned immediately after being sent. The message system ensures reliable delivery of messages. The message publisher just publishes the message to MQ and doesn't care who gets it, and the message consumer just gets the message from MQ regardless of who publishes it. In this way, neither the publisher nor the user needs to know the existence of the other party.

2. What are the commonly used message queues? RabbitMQ, RocketMQ, ActiveMQ, Kafka, ZeroMQ, MetaMq.

Even now some NoSQL can also be used as message queues, such as Redis.

3. What are the usage scenarios of message queue? Asynchronous processing application decoupling traffic peak shaving 4. Use cases Large companies will have their own log analysis system. How is the log system implemented?

Illustration: When a user accesses an application, we need to record the user's operation records and system exception logs. The conventional approach is to save the logs generated by the system to the server disk and start scheduled tasks in the server. , regularly transfer the log information from the disk to mq (producer), and also regularly extract the messages from mq and store them in the corresponding database, such as ElasticSearch or Hive.

5. How to install RabbitMQ? The above case introduces a usage scenario of MQ. I am using RabbitMQ as an example here. Kafka may be used in real projects.

First install brew (mac as an example)

/usr/bin/ruby -e "$(curl -fsSL /Homebrew/install/master/install)"Install RabbitMQ

brew install rabbitmq to run RabbitMQ

Go to /usr/local/Cellar/rabbitmq/3.7.7 and execute

sbin/rabbitmq-server to start the plug-in

Enter /usr/local/Cellar/rabbitmq/3.7.7/sbin

./rabbitmq-plugins enable rabbitmq_management login management interface

Open the browser and enter: http ://localhost: 15672, RabbitMQ defaults to 15672 port six. Nodejs operates RabbitMQ

Several corresponding Node SDKs can be found online, and amqplib is recommended here

1. Producer

/**

* Encapsulation of RabbitMQ

*/

let amqp = require('amqplib');

class RabbitMQ {

constructor() {

this.hosts = [];

this.index = 0;

this.length = this.hosts.length;

this.open = amqp.connect(this.hosts[this.index]);

}

sendQueueMsg(queueName, msg, errCallBack) {

let self = this;

self.open

.then(function (conn) {

return conn.createChannel();

})

.then(function (channel) {

return channel.assertQueue(queueName).then (function (ok) {

return channel.sendToQueue(queueName, new Buffer(msg), {

persistent: true

});

})

.then(function (data) {

if (data) {

errCallBack amp; errCallBack("success" );

channel.close();

}

})

.catch(function () {

setTimeout(() =gt; {

if (channel) {

channel.close();

}

}, 500)

});

})

p>

.catch(function () {

let num = self.index;

if (num lt; = self.length - 1) {

self.open = amqp.connect(self.hosts[num]);

} else {

self.index == 0;

}

});

}

}2. Consumer

/**

* Yes RabbitMQ encapsulation

*/

let amqp = require('amqplib');

class RabbitMQ {

constructor() {

this.open = amqp.connect(this.hosts[this.index]);

}

receiveQueueMsg(queueName, receiveCallBack, errCallBack) {

let self = this;

self.open

.then(function (conn) {

return conn.createChannel();

})

.then(function (channel) {

return channel.assertQueue(queueName)

.then(function ( ok) {

return channel.consume(queueName, function (msg) {

if (msg !== null) {

let data = msg. content.toString();

channel.ack(msg);

receiveCallBack amp; amp; receiveCallBack(data);

}

})

.finally(function () {

setTimeout(() =gt; {

if (channel) {

channel.close();

}

}, 500)

});

})

})

.catch(function () {

let num = self.index;

if (num lt; = self.length - 1) {

self.open = amqp.connect(self.hosts[num]);

} else {

self.index = 0;

self.open = amqp.connect(self.hosts[0]);

}

});

}3. Send a message to MQ through the producer and create a queue

let mq =

new RabbitMQ();

mq.sendQueueMsg('testQueue', 'my first message', (error) =gt; {

console.log(error)

}) After execution, we opened the management platform and found that RabbbitMQ had received a message:

And RabbbitMQ added a new queue testQueue

4. Get the message of the specified queue

let mq = new RabbitMQ();

mq.receiveQueueMsg('testQueue', (msg) =gt; {

console.log(msg)

})// Output result: my first message Open the RabbitMQ management platform at this time, and the number of messages has become 0

In summary: We briefly talked about the message queue and RabbitMQ related Some knowledge and how we produce and consume messages through nodejs.