在Node.js中实现消费rabbitmq队列消息示例

前端  /  管理员 发布于 11个月前   294

在Node.js中实现消费rabbitmq队列消息功能。

示例代码：

index.js

var amqp = require('amqplib/callback_api');
const MyConsume = require('./MyConsume');
amqp.connect('amqp://name:password!@localhost:5672/vhost', function (error0, connection) {
    if (error0) {
        throw error0;
    }
    connection.createChannel(function (error1, channel) {
        if (error1) {
            throw error1;
        }
        var exchangeName = 'my_topic'; // 你的 exchange 名称
        var exchangeType = 'topic'; // exchange 类型，可以是 direct、fanout、topic 等
        var routingKey = 'my_routing_key'; // 你的 routing key
        var queueName = 'my_queue'; // 你的队列名称
        channel.assertExchange(exchangeName, exchangeType, { durable: true }, function(error2, ok) {
            if (error2) {
                throw error2;
            }
            console.log('Exchange ' + exchangeName + ' is ready');
            // 声明一个队列
            channel.assertQueue(queueName, { durable: true  }, function(error3, q) {
                if (error3) {
                    throw error3;
                }
                console.log('Queue ' +queueName + ' is created');
                // 将队列绑定到 exchange，并指定 routing key
                channel.bindQueue(queueName, exchangeName, routingKey);
                console.log('Queue ' + queueName + ' is bound to Exchange ' + exchangeName + ' with routing key ' + routingKey);
                // 在这里开始消费消息
                channel.consume(queueName, function(msg) {
                  MyConsume.handleMessage(msg,channel);
                }, {
                    noAck: false
                });
            });
        });
    });
});

MyConsume.js

const { UserModel } = require('./UserModel');
function handleMessage(msg,channel) {
    UserModel.create({
        user_name: 'Example Bookmark',
        url: 'http://www.example.com',
        type_id: 1,
        order: 1,
        is_recommend: true,
        status: true
      }).then(bookmark => {
        console.log('New bookmark created:');
        console.log(new Date());
      }).catch(error => {
        console.error('Error creating bookmark:', error);
      });
    // 在这里可以添加你的其他处理逻辑
    channel.ack(msg)
}
module.exports = {
    handleMessage: handleMessage
};

使用 Mysql 连接池，开 10 个 mysql 连接，消费 3 万 rabbitmq 消息，

每条消息的逻辑是往 mysql 数据表写入一条数据，30 秒写入完.

以上 nodejs 代码的写法我用 php 实现不到。

注意：
需要处理完消息手动确认消息.