Message retardé dans RabbitMQ

Il existe deux approches que vous pouvez essayer:

L'ancienne Approche: Définir le TTL(time to live) de l'en-tête de chaque message/file d'attente(la politique) et ensuite introduire une DLQ de le gérer. une fois la durée de vie a expiré vos messages vont passer de DLQ à la file d'attente principale, de sorte que votre auditeur peut la traiter.

Dernière Approche: Récemment RabbitMQ est venu avec RabbitMQ Message Retardé Plugin , à l'aide de laquelle vous pouvez atteindre le même et ce support de plugins disponibles depuis RabbitMQ-3.5.8.

Vous pouvez déclarer un échange avec le type de x-retard de message et publier des messages avec la coutume d'en-tête x-délai exprimé en millisecondes, un temps de retard pour le message. Le message sera envoyé à la files d'attente respectives, après x-retard millisecondes

byte[] messageBodyBytes = "delayed payload".getBytes("UTF-8");
Map<String, Object> headers = new HashMap<String, Object>();
headers.put("x-delay", 5000);
AMQP.BasicProperties.Builder props = new 
AMQP.BasicProperties.Builder().headers(headers);
channel.basicPublish("my-exchange", "", props.build(), messageBodyBytes);

Plus ici: git

Merci à Norman réponse, j'ai pu mettre en œuvre dans NodeJS.

Tout est assez clair dans le code.
Espérons qu'il permettra de sauver quelqu'un de l'époque.

var ch = channel;
ch.assertExchange("my_intermediate_exchange", 'fanout', {durable: false});
ch.assertExchange("my_final_delayed_exchange", 'fanout', {durable: false});

//setup intermediate queue which will never be listened.
//all messages are TTLed so when they are "dead", they come to another exchange
ch.assertQueue("my_intermediate_queue", {
      deadLetterExchange: "my_final_delayed_exchange",
      messageTtl: 5000, //5sec
}, function (err, q) {
      ch.bindQueue(q.queue, "my_intermediate_exchange", '');
});

ch.assertQueue("my_final_delayed_queue", {}, function (err, q) {
      ch.bindQueue(q.queue, "my_final_delayed_exchange", '');

      ch.consume(q.queue, function (msg) {
          console.log("delayed - [x] %s", msg.content.toString());
      }, {noAck: true});
});

Il ressemble ce blog décrit l'utilisation de la lettre morte, l'échange et le message ttl à faire quelque chose de similaire.

Le code ci-dessous utilise CoffeeScript et Node.JS pour accéder à Lapin et de mettre en œuvre quelque chose de similaire.

amqp   = require 'amqp'
events = require 'events'
em     = new events.EventEmitter()
conn   = amqp.createConnection()

key = "send.later.#{new Date().getTime()}"
conn.on 'ready', ->
  conn.queue key, {
    arguments:{
      "x-dead-letter-exchange":"immediate"
    , "x-message-ttl": 5000
    , "x-expires": 6000
    }
  }, ->
    conn.publish key, {v:1}, {contentType:'application/json'}

  conn.exchange 'immediate'

  conn.queue 'right.now.queue', {
      autoDelete: false
    , durable: true
  }, (q) ->
    q.bind('immediate', 'right.now.queue')
    q.subscribe (msg, headers, deliveryInfo) ->
      console.log msg
      console.log headers

Supposons que vous ayez un contrôle sur la consommation, vous pouvez obtenir le retardement sur le consommateur, comme cette??:

Si nous sommes sûrs que le n-ième message dans la file d'attente a toujours un petit retard que les n+1) ème message (ceci peut être vrai pour de nombreux cas d'utilisation): Le producteur envoie timeInformation dans la tâche de transmettre l'heure à laquelle ce travail doit être exécuté (currentTime + delay). Le consommateur:

1) Lit le scheduledTime de la tâche

2) si currentTime > scheduledTime aller de l'avant.

D'autre delay = scheduledTime - currentTime

dormir pendant le temps indiqué par le retard

Le consommateur est toujours configuré avec une simultanéité paramètre. Donc, les autres messages juste attendre dans la file d'attente jusqu'à ce qu'un consommateur termine le travail. Donc, cette solution pourrait seulement fonctionne bien mais il a l'air gêné, surtout pour les gros temps de retard.