Spring boot und rabbitmq integration, wie man bei komponentenausfall wiederherstellen?

Question

Ich habe eine Spring Rest App mit Hase als Messaging Middleware. Wenn die App eine eingehende Ruheanforderung annimmt, antwortet sie dem Benutzer, während sie eine zusätzliche Verarbeitung asynchron durchführt, eine neue Nachricht erzeugt und sie auf Hase setzt. Von dort lesen einige Verbraucher die Nachricht und liefern sie an das externe System. Nun, es gibt einen Fall, dass der Kaninchen-Server selbst nicht verfügbar ist. Um diese Art von Fehlern zu umgehen, ich habe eine blockierende Warteschlange (natürlich mit einer einstellbaren Größe)

@Bean 
public TaskExecutor taskExecutor() { 
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); 
    executor.setCorePoolSize(5); 
    executor.setMaxPoolSize(10); 
    executor.setQueueCapacity(25); 

    return executor; 
} 

Die Warteschlange der Lage, einige Nachrichten zu verarbeiten, bis Kaninchen online geht. Wenn es jedoch zu lange dauert und die Anforderungen die Warteschlangengröße überschreiten, wird es fehlschlagen. Jetzt ist die Frage, gibt es eine Möglichkeit, dass ich das verbessern und effizienter machen kann? Was ist die beste Praxis der Branche? Gibt es Muster für dieses Szenario?

Answer 1

Ich verwendete Frühling wiederholen. Das reicht für die Anforderung ziemlich aus. Wie ich es wiederholen kann, konfiguriere Wiederholungsintervalle, maximale Versuche usw. und implementiere die benutzerdefinierte Wiederherstellungsfunktion.

@Async 
    @Retryable(value = { AmqpException.class }, 
      maxAttempts=10, backoff=@Backoff(delay=100, maxDelay=300000,multiplier = 2)) 
    public void sendEmailMessageToQueue(SimpleMailMessage email){ 
     log.info("Sending message to queue"+Thread.currentThread().getName()); 
     try { 
      rabbitTemplate.convertAndSend(queueName, email); 
     }catch (AmqpException e){ 
      log.log(Level.WARNING,"Could not send message to queue, will retry.",e); 
     } 
    } 

    @Recover 
    public void connectionException(AmqpException e) { 
     log.log(Level.SEVERE,"Could not send message to queue, will save the message in db instead."+Thread.currentThread().getName(),e); 
    }