Rabbit MQ & NodeJS & Protobuf
一、NodeJS Rabbit MQ 客户端封装
利用第三方库amqlib封装了一个简单的收发客户端。
// rabbit.js
/**
* 对RabbitMQ的封装
*/
let amqp = require('amqplib');
class RabbitMQ {
constructor() {
this.hosts = ['amqp://192.168.94.105'];
this.index = 0;
this.length = this.hosts.length;
this.open = amqp.connect({
hostname:'192.168.94.105',
port:5672,
username:"lixin",
password:"lixin"
});
}
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();
let data = msg.content;//binary data
channel.ack(msg);
receiveCallBack && receiveCallBack(data);
}
}).finally(function () {
setTimeout(() => {
if (channel) {
channel.close();
}
}, 500)
});
})
}).catch(function () {
let num = self.index++;
if (num <= self.length - 1) {
self.open = amqp.connect(self.hosts[num]);
} else {
self.index = 0;
self.open = amqp.connect(self.hosts[0]);
}
});
}
//msg 二进制流
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, (msg), {persistent: true});
}).then(function (data) {
if (data) {
errCallBack && errCallBack("success");
channel.close();
}
}).catch(function () {
setTimeout(() => {
if (channel) {
channel.close();
}
}, 500)
});}).catch(function () {
let num = self.index++;
if (num <= self.length - 1) {
self.open = amqp.connect(self.hosts[num]);
} else {
self.index = 0;
}
});
}
}
exports.RabbitMQ = RabbitMQ;
二、结合Protobuf
proto文件生成js
- proto文件定义
syntax = "proto3";
message Test{
string label = 1;
int32 type = 2;
int64 reps = 3;
}
- 利用protoc程序在终端输入以下命令生成test_pb.js文件
.\protoc.exe --js_out=import_style=commonjs,binary:. .\test.proto
- 生产者
//producer.js
let RabbitMQ = require('./rabbit').RabbitMQ;
let mq = new RabbitMQ();
var P = require('./test_pb');
var test = new P.Test();
test.setType(1);
test.setLabel("karl");
test.setReps(123);
var bit8Arr = test.serializeBinary();
setInterval(
() => {
mq.sendQueueMsg('testQueue', Buffer.from(bit8Arr), (error) => {
console.log(error);
})},10000);
- 消费者
//consumer.js
pb = require('./test_pb.js');
let RabbitMQ = require('./rabbit').RabbitMQ;
let mq = new RabbitMQ();
setInterval(()=>{
mq.receiveQueueMsg('testQueue',(msg) =>
{
var out = pb.Test.deserializeBinary(msg);
console.log("Test{",out.getLabel(),":",out.getType(),":",out.getReps(),"}");
})},500);
- 验证
分别在两个终端运行 node consumer.js node producer.js 可以在消费者终端看到运行结果
Test{ karl : 1 : 123 }
Test{ karl : 1 : 123 }
可以看到已经正确消费protobuf消息。而对于json的传输也一样，相当于传送字符串，前一篇文章可以直接解出来。
三、总结
Rabbit MQ就是收发二进制流，至于怎么解析，看具体的协议的实现。
