AMQP Consume Messages

Consumer konsumieren aus Warteschlangen. Um Nachrichten zu konsumieren, muss es eine Warteschlange geben. Wenn ein neuer Consumer hinzugefügt wird, beginnen die Zustellungen sofort, sofern bereits Nachrichten in der Warteschlange bereitstehen.

Die Ziel-Queue kann zum Zeitpunkt der Konsumentenregistrierung leer sein. In diesem Fall finden die ersten Zustellungen statt, wenn neue Nachrichten in die Queue gestellt werden.

 

Das Konsumieren von Nachrichten ist eine asynchrone Aufgabe, was bedeutet, dass jedes Mal, wenn eine neue Nachricht an die Consumer-Queue zugestellt werden kann, sie automatisch vom Server an den Client gepusht wird. Sie können eine alternative Methode zum synchronen Empfangen von Nachrichten lesen.

 

Consume

Die Methode Consume erstellt einen neuen Consumer in der Warteschlange, und jedes Mal, wenn es eine neue Nachricht gibt, wird diese automatisch an den Consumer-Client übermittelt.

 

Die Methode hat die folgenden Argumente:

 

• ChannelName: ist der Name des Channels (muss geöffnet sein, bevor diese Methode aufgerufen wird).
• QueueName: es ist der Name der Queue, darf nicht länger als 255 Zeichen sein und nicht mit "amq." beginnen (außer wenn der Passive-Parameter true ist).
• ConsumerTag: Es ist der Name des Consumers und muss eindeutig sein. Wenn er nicht gesetzt ist, erstellt der Server einen neuen.
• NoLocal: Wenn true, bedeutet dies, dass der Verbraucher niemals Nachrichten konsumiert, die auf demselben Kanal veröffentlicht wurden.
• NoAck: wenn true, bedeutet dies, dass der Server keine Bestätigung für jede zugestellte Nachricht erwartet.
• Exclusive: wenn true, verhindert dies, dass andere Consumer Nachrichten aus dieser Queue konsumieren.
• NoWait: wenn true, sendet der Server keine Bestätigung an den Client.
• Arguments: Zeichenfolge, die benutzerdefinierte Argumente enthält, die Werte müssen als JSON-Zeichenfolge übergeben werden, Beispiel: {"x-dead-letter-exchange":"my-dlx"}.

 

Die Nachrichten werden im Ereignis OnAMQPBasicDeliver zugestellt.

 

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;

Ein synchroner Aufruf kann einfach durch Aufrufen der Methode ConsumeEx erfolgen; diese Methode gibt true zurück, wenn der Consumer erstellt wurde, und false, wenn keine Bestätigung vom Server eingegangen ist.

 

Consume abbrechen

Diese Methode wird verwendet, um eine vorhandene Konsumenten-Queue zu stornieren.

 

Die Methode hat die folgenden Argumente:

 

• ChannelName: ist der Name des Channels (muss geöffnet sein, bevor diese Methode aufgerufen wird).
• ConsumerTag: Es ist der Name des Verbrauchers.
• NoWait: wenn true, sendet der Server keine Bestätigung an den Client.

 

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

Ein synchroner Aufruf kann einfach durch Aufruf der Methode CancelConsumeEx erfolgen; diese Methode gibt true zurück, wenn der Consumer abgebrochen wurde, und false, wenn keine Bestätigung vom Server eingetroffen ist.