どうやって使うの?
では実際にLazy Queuesを設定したキューを定義してやり、キューに対してメッセージを送る処理を記述してみます。
まずはLazy Queuesを使わない、通常のキューを用いて以下の単純なPub-Subモデルのシステムを記述します。実装に当たり、Ruby(v2.2.2)とRuby版RabbitMQライブラリBunnyを用います。
![system model](http://cz-cdn.shoeisha.jp/static/images/article/9288/9288_model_s.png)
通常のキュー
まずはじめに、メッセージの送信処理を記述します。以下のコードはRabbitMQサーバ「mq-server」に対して、1KBのメッセージ送る処理になります。
require 'bunny' c = Bunny.new(:host => 'mq-server', :user => 'tsuchinoko', :password => 'passwd') c.start ch = c.create_channel q = ch.queue('basic-queue') begin loop do q.publish('a' * 1024) end rescue Exception => _ ch.close c.close end
コードの中身を簡単に解説すると、3行目で設定したRabbitMQサーバの情報に基づいて、4行目でコネクションを確立させます。次に6行目でChannelと呼ばれるオブジェクトを生成します。ChannelはAMQPで規定されているメッセージブローカー内部での論理的接続で、一つのTCPコネクションで複数のChannelを定義させることができます。またキュー、エクスチェンジの設定・状態はChannelごとに独立します。なのでChannelに対する操作はスレッドセーフであり、安全に並列処理が行えます。
そして7行目で「basic-queue」と名前付けしたキューを生成しています。10-12行目で、このキューに対してサイズ1KBのメッセージを送り続けます。
続いて、受信処理を記述します。
require 'bunny' c = Bunny.new(:host => 'mq-server', :user => 'tsuchinoko', :password => 'passwd') c.start ch = c.create_channel q = ch.queue('default-queue') begin q.subscribe(:block => true) do |delivery_info, properties, data| puts data end rescue Exception => _ ch.close c.close end
7行目まではメッセージ送信処理と同じです。メッセージの受信は10行目のsubscribeで行われます。サーバのキュー「basic-queue」にメッセージが存在すると、ブロック内部の処理が呼ばれます。ブロックで受け取っている3つのパラメータはそれぞれ以下のとおりです。
- delivery_info:どういったアプリケーションから、どういった経路で送られたか、といったメッセージ転送処理に関する情報
- properties:データタイプや永続化されているか、といったメッセージ自体のメタデータ情報
- data:キューに格納されたメッセージの本体
Lazy Queuesを設定したキュー
では、次にLazy Queuesを用いた送信処理を記述します。
Lazy Queuesは、キューを宣言する際にパラメータ「x-queue-mode」に「lazy」を指定することでLazy Queuesを設定できます(デフォルトではLazy Queues非設定の「default」が指定されます)。
以下は、Lazy Queuesを設定したキューを用いたメッセージの送信処理になります。
require 'bunny' c = Bunny.new(:host => 'mq-server', :user => 'tsuchinoko', :password => 'passwd') c.start ch = c.create_channel q = ch.queue('lazy-queue', :arguments => {'x-queue-mode' => 'lazy'}) begin loop do q.publish('a' * 1024) end rescue Exception => _ ch.close c.close end
先に示した、通常のキューの実装との差分は以下のとおりです。
--- basic_queues/sender.rb 2016-02-14 15:25:51.000000000 +0900 +++ lazy_queues/sender.rb 2016-02-14 23:35:36.000000000 +0900 @@ -6,7 +6,7 @@ c.start ch = c.create_channel -q = ch.queue('basic-queue') +q = ch.queue('lazy-queue', :arguments => {'x-queue-mode' => 'lazy'}) begin loop do
RabbitMQでは、作成済みのキューに対してパラメータの変更を行うことはできないため、通常のキューとは別の名前「lazy-queue」を設定しています。すでに作成されているキューと同名のキューに対してLazy Queuesを設定する場合、deleteメソッドによって当該キューを削除してから、再定義してやる必要があります。
同様に受信処理についても、Lazy Queuesを用いた場合とそうでない場合の差分を以下に示します。
--- basic_queues/receive.rb 2016-02-14 23:36:04.000000000 +0900 +++ lazy_queues/receive.rb 2016-02-14 23:36:28.000000000 +0900 @@ -6,7 +6,7 @@ c.start ch = c.create_channel -q = ch.queue('default-queue') +q = ch.queue('lazy-queue', :arguments => {'x-queue-mode' => 'lazy'}) begin q.subscribe(:block => true) do |delivery_info, properties, data|
では実際にこれら2つの処理を動かして、RabbitMQサーバの振る舞いの違いを見てみましょう。