AMQP Consume Messages

コンシューマーはキューから消費します。メッセージを消費するにはキューが必要です。新しいコンシューマーが追加されると、キューにすでにメッセージが準備されている場合、配信は即座に開始されます。

コンシューマー登録時にターゲットキューが空の場合があります。その場合、新しいメッセージがエンキューされたときに最初の配信が行われます。

 

メッセージの消費は非同期タスクです。これは、コンシューマーキューに新しいメッセージを配信できるたびに、サーバーからクライアントに自動的にプッシュされることを意味します。同期的にメッセージを受信する代替方法を参照してください。

 

消費

Consume メソッドはキューに新しいコンシューマーを作成し、新しいメッセージがあるたびにそれがコンシューマークライアントに自動的に配信されます。

 

このメソッドには以下の引数があります:

 

• ChannelName:チャンネルの名前（このメソッドを呼び出す前にオープンしておく必要があります）。
• QueueName: キューの名前。255 文字以内で、"amq." で始まってはなりません（passive パラメータが true の場合を除く）。
• ConsumerTag: コンシューマーの名前で、一意である必要があります。設定されていない場合、サーバーが新しい名前を作成します。
• NoLocal: trueの場合、コンシューマーは同じチャンネルに発行されたメッセージを決して消費しないことを意味します。
• NoAck: trueの場合、サーバーは配信されたすべてのメッセージに対して確認応答を期待しないことを意味します。
• Exclusive: true の場合、他のコンシューマーがこのキューからメッセージを消費することを防ぎます。
• NoWait: True の場合、サーバーはクライアントに確認応答を送信しません。
• Arguments: カスタム引数を含む文字列。値はJSON文字列として渡す必要があります。例: {"x-dead-letter-exchange":"my-dlx"}。

 

メッセージは OnAMQPBasicDeliver イベントで配信されます。

 

AMQP.Consume('channel_name', 'queue_name', 'consumer_tag');
 
procedure OnAMQPBasicDeliver(Sender: TObject;
  const aChannel: string;
  const aDeliver: TsgcAMQPFramePayload_Method_BasicDeliver;
  const aContent: TsgcAMQPMessageContent);
begin
  DoLog('#AMQP_basic_deliver: ' + aChannel + ' ' + aDeliver.ConsumerTag + ' ' +
    ' ' + aContent.Body.AsString);
end;

ConsumeEx メソッドを呼び出すことで同期呼び出しも可能です。このメソッドはコンシューマーが作成された場合に true を返し、サーバーから確認が届かなかった場合は false を返します。

 

Cancel Consume

このメソッドは、既存のコンシューマーキューをキャンセルするために使用されます。

 

このメソッドには以下の引数があります:

 

• ChannelName:チャンネルの名前（このメソッドを呼び出す前にオープンしておく必要があります）。
• ConsumerTag: コンシューマーの名前です。
• NoWait: True の場合、サーバーはクライアントに確認応答を送信しません。

 

AMQP.CancelConsume('channel_name', 'consumer_tag');
 
procedure OnAMQPBasicCancelConsume(Sender: TObject; const aChannel: string; const aConsumerTag);
begin
  DoLog('#AMQP_basic_cancel_consume: ' + aChannel + ' ' + aConsumerTag);
end;

メソッド CancelConsumeEx を呼び出すだけで同期呼び出しも可能です。このメソッドはコンシューマーがキャンセルされた場合に true を返し、サーバーからの確認が届かない場合は false を返します。