AMQP 消费消息

消费者从队列中消费消息。为了消费消息，必须存在一个队列。添加新消费者后，假设队列中已有就绪消息，投递将立即开始。

消费者注册时目标队列可以为空。在这种情况下，当新消息入队后，首次投递才会发生。

 

消费消息是一项异步任务，这意味着每当有新消息可以投递到消费者队列时，服务器会自动将其推送给客户端。您可以阅读另一种同步接收消息的方法。

 

消费

方法 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。

 

取消消费

此方法用于取消现有的消费者队列。

 

该方法具有以下参数：

 

• 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。