nodejs队列实现amqplib,rabbitmq

2020-12-07 16:14:22

参考到地址 对于amqplib的使用心得

     最近在nodejs使用了amqplib--rabbitmq的nodejs客户端。封装在了express中,先来代码。

var amqp = require('amqplib/callback_api');
var config=require('../config/config');
var log=require('../util/loghelp');
function fail(err, conn) {
    log.error(err);
    if (conn) conn.close();
}
exports.StartConsumer=function (action,qname) {
    function on_connect(err, conn) {
        if (err !== null) return fail(err);
        function on_channel_open(err, ch) {
            ch.assertQueue(qname, {durable: true}, function(err, ok) {
                if (err !== null) return bail(err, conn);
                ch.consume(qname, function(msg) {

                    log.info(`Received ${msg.content.toString()},start process`);
                    action(JSON.parse(msg.content))
                        .then(d=> {
                            log.info("mq 处理成功,确认");ch.ack(msg)
                        }
                           )
                        .catch(err=>
                        ch.nack(msg));
                }, {noAck: false} );
            });
        }
        conn.createChannel(on_channel_open);
    }
    amqp.connect(config.amqp.url,on_connect);
};

exports.enqueue=function (data,qname) {
    function on_connect(err, conn) {
        if (err !== null) return bail(err);

        function on_channel_open(err, ch) {
            if (err !== null) return bail(err, conn);
            ch.assertQueue(qname, {durable: true}, function(err, ok) {
                if (err !== null) return bail(err, conn);
                var msg=JSON.stringify(data);
                ch.sendToQueue(qname, new Buffer(msg));
                log.info(`mq send ${msg}`);
                ch.close(function() { conn.close(); });
            });
        }
        conn.createChannel(on_channel_open);
    }
    amqp.connect(config.amqp.url,on_connect);
};


其中StartConsumer 会在项目启动时启动,在整个生命周期中一直保持监听状态,在程序结束时mq的链接关闭。需要注意的是 noAck 这个参数,当为false是表示消息出队后不会自动删除,如果设置成true,则无论消息处理成功与否此消息会被删除。注意到在消息不成功是,调用了ch.nack(msg)),此方法是将消息重新入队。

   而enqueue 则是消息入队列后连接立刻关闭,以免占用资源。


  • 2020-01-06 23:02:42

    Updating Homebrew... 更新了镜像依然卡死

    使用brew install [软件包]安装软件包时,卡在Updating Homebrew... 或输入`brew update`更新brew,半天没反应.产生原因一般是在国内访问官方 更新源获取资源太慢,解决方案可以采用更换国内镜像更新源.

  • 2020-01-07 10:06:00

    vue重定向beforeRouterEnter与replace的使用

    一个登录页面,只有第一次才会出现,一旦出现过,以后再也不会出现,即使是输入该页面的url也不会跳转到该页面,只会跳转到指定的其他页面;

  • 2020-01-07 10:14:07

    vue强制刷新组件 销毁和重建

    很多时候,通过重置数据将页面重置时,子组件可以提供重置的方法,或者提供props重置自己的状态。但是相对麻烦,那可以使用强制刷新来实现刷新组件。

  • 2020-01-07 10:17:42

    vue钩子函数beforeRouteUpdate没有反应

    由于会渲染同样的 Foo 组件,因此组件实例会被复用。而这个钩子就会在这个情况下被调用。 可以访问组件实例 `this`

  • 2020-01-08 13:23:24

    vue 中使用eventbus

    为了提高组件的独立性和重用性,父组件会通过props向下传数据给子组件,当子组件又事情要告诉父组件时用通过$emit事件告诉父组件,如此确保每个组件都是独立在相对隔离的环境中运行,可以大幅度提高组件的可维护性

  • 2020-01-08 13:30:30

    vue中eventbus被多次触发(vue中使用eventbus踩过的坑)

    一开始的需求是这样子的,我为了实现两个页面组件之间的数据传递,假设我有页面A,点击页面A上的某一个按钮之后,页面会自动跳转到页面B,同时我希望将页面A上的某一些参数携带过去给页面B。 然后我就想,这不就是不同组件之间的数据传递问题而已吗?直接用bus 巴士事件来传递数据不就行了吗。于是,我就很愉快地进行了。关于vue中的eventbus的使用,我之前在一篇vue中的数据传递中有提到过。