NodeJS によるメッセージキュー RabbitMQ の操作方法の分析-jsチュートリアル-php.cn

NodeJS によるメッセージキュー RabbitMQ の操作方法の分析

不言

リリース： 2018-07-14 16:03:37

オリジナル

2801 人が閲覧しました

この記事では、NodeJS がどのようにメッセージキューを操作するかを主に紹介します。これには、必要な参考値があります。

1.

メッセージとは、アプリケーション間で送信されるデータを指します。メッセージは、テキスト文字列のみを含む非常に単純な場合もあれば、埋め込みオブジェクトを含む場合があるより複雑な場合もあります。

メッセージキュー (Message Queue) は、メッセージを送信後すぐに返すことができるメッセージシステムです。メッセージ発行者はメッセージを MQ に発行するだけで、誰がメッセージを取得するかは気にしません。また、メッセージコンシューマーは、誰が発行するかに関係なく、MQ からメッセージを取得するだけです。これにより、発行者もユーザーも相手の存在を知る必要がなくなります。

2. 一般的に使用されるメッセージキューとは何ですか?

RabbitMQ、RocketMQ、ActiveMQ、Kafka、ZeroMQ、MetaMq。

現在でも、Redis などの一部の NoSQL はメッセージキューとして使用できます。

3. メッセージキューの使用シナリオは何ですか?

非同期処理
アプリケーションのデカップリング
トラフィックピークの削減

4. ユースケース

大規模企業は独自のログ分析システムをどのように実装していますか?

図: ユーザーがアプリケーションにアクセスするとき、ユーザーの操作記録とシステム例外ログを記録する必要があります。従来のアプローチは、システムによって生成されたログをサーバーディスクに保存し、タイミングを有効にすることです。サーバーのタスクは、ログ情報をディスクから mq (プロデューサー) に定期的に転送し、mq からメッセージを定期的に抽出して、ElasticSearch や Hive などの対応するデータベースに保存することです。

5. RabbitMQ をインストールするには?

上記のケースでは、実際のプロジェクトで使用できる例として RabbitMQ を使用しています。

最初にbrewをインストールします(例としてMac)

/usr/bin/ruby -e "$(curl -fsSL https://raw.githubusercontent.com/Homebrew/install/master/install)"

ログイン後にコピー

RabbitMQをインストールします

brew install rabbitmq

ログイン後にコピー

RabbitMQを実行します

/usr/local/Cellar/rabbitmq/3.7.7に移動し、実行します

sbin/rabbitmq-server

ログイン後にコピー

プラグインを開始します

Go /usr/local/Cellar/rabbitmq/3.7.7/sbin

./rabbitmq-plugins enable rabbitmq_management

ログイン後にコピー

管理インターフェイスにログインします

ブラウザを開いて次のように入力します: http://localhost:15672、RabbitMQ のデフォルト 15672 Nodejs は RabbitMQ を動作させます

対応するノード SDK がいくつか見つかりました。ここでは amqplib

1. プロデューサー

/**
 * 对RabbitMQ的封装
 */
let amqp = require(&#39;amqplib&#39;);

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 && 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;
                }
            });
    }
}

ログイン後にコピー

2. Consumer

/**
 * 对RabbitMQ的封装
 */
let amqp = require(&#39;amqplib&#39;);

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 && 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]);
                }
            });
    }

ログイン後にコピー

3. プロデューサーを通じて MQ にメッセージを送信し、キュー

let mq = new RabbitMQ();
mq.sendQueueMsg(&#39;testQueue&#39;, &#39;my first message&#39;, (error) => {
    console.log(error)
})

ログイン後にコピー

を作成します。実行後、管理プラットフォームを開くと、RabbitMQ がメッセージ

を受信していることがわかり、RabbitMQ は新しいキュー testQueue

4 指定されたキューのメッセージを取得します。

この時点で、RabbitMQ 管理プラットフォームを開くと、メッセージの数が 0 に変わりました

要約すると、メッセージキューと RabbitMQ に関連する知識と、nodejs を介してメッセージを生成および消費する方法について簡単に説明しました。。

上記がこの記事の全内容です。その他の関連コンテンツについては、PHP 中国語 Web サイトをご覧ください。

NodeJS によるメッセージキュー RabbitMQ の操作方法の分析

1.

2. 一般的に使用されるメッセージ キューとは何ですか?

3. メッセージキューの使用シナリオは何ですか?

非同期処理

アプリケーションのデカップリング

トラフィックピークの削減

4. ユースケース

5. RabbitMQ をインストールするには?

2. 一般的に使用されるメッセージキューとは何ですか?