YangHuCost/lib/rabbitmq/RabbitMQ.js

/**
 * Created by zhang on 2019/5/5.
 */
let amqp = require('amqplib');
class RabbitMQ{
    constructor(connect){
        this.connect = connect;
    }

    static async connection(URL){
        if(!RabbitMQ.instance){
            RabbitMQ.instance =  new RabbitMQ(await amqp.connect(URL)) //'amqp://13726297388:123456@qa.smartcost.com.cn:5672'
        }
        return RabbitMQ.instance
    }

    static async sendMessage(queue,msg,durable=false){
        if(RabbitMQ.instance){
            let ch = await RabbitMQ.instance.connect.createChannel();
            try {
                await ch.assertQueue(queue, {durable: durable});
                ch.sendToQueue(queue,Buffer.from(msg),{persistent: durable});
                console.log(" Sent %s to :"+queue, msg);
                ch.close();
            }catch (e){
                console.log(e);
                ch.close();
            }

        }
    }

    static async receiveMessage(queue,callback,durable=false){
        if(RabbitMQ.instance){
            let ch = await RabbitMQ.instance.connect.createChannel();
            try {
                await ch.assertQueue(queue, {durable: durable});
                ch.prefetch(1);
                console.log(" [*] Waiting for messages in %s. To exit press CTRL+C", queue);
                ch.consume(queue, function(msg) {
                    console.log(" [x] Received %s", msg.content.toString());
                    if(durable == true) ch.ack(msg);
                    if(callback) callback(msg)
                }, {
                    noAck: !durable
                });
            }catch (e){
                console.log(e);
                ch.close();
            }
        }
    }
}

export default RabbitMQ;