Spring Rabbitmq DLQ su Mished MASSAGE_ID

Question

Attualmente ho configurato il mio contenitore di ascolto del messaggio per utilizzare la molla-riprova per la gestione dei tentativi di gestione, ma quando qualcuno invia un messaggio senza dare un messaggio-ID messaggio, il messaggio listener si arresta.Questo comportamento può essere modificati in modo da mettere il messaggio sulla coda della lettera morta invece di fermare l'ascoltatore?

La mia configurazione per riprovare è la seguente:

 @Bean
 public StatefulRetryOperationsInterceptor retryInterceptor() {
   StatefulRetryOperationsInterceptorFactoryBean f = new     
   StatefulRetryOperationsInterceptorFactoryBean();
   f.setRetryOperations(retryTemplate());
   f.setMessageRecoverer(new RejectAndDontRequeueRecoverer());

   return f.getObject();
}

private RetryOperations retryTemplate() {
  RetryTemplate retryTemplate = new RetryTemplate();
  ExponentialRandomBackOffPolicy backOffPolicy = new ExponentialRandomBackOffPolicy();

  backOffPolicy.setInitialInterval(50);
  backOffPolicy.setMultiplier(1.5);
  backOffPolicy.setMaxInterval(1000);
  retryTemplate.setBackOffPolicy(backOffPolicy);

  SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
  retryPolicy.setMaxAttempts(10);
  retryTemplate.setRetryPolicy(retryPolicy);

  return retryTemplate;
}

.

E ottengo la seguente eccezione:

2014-08-01 08:50:27,858 [taskExecutor<OmittedForPrivacy>-2] WARN  mqp.rabbit.listener.SimpleMessageListenerContainer - Execution of Rabbit message listener failed, and no ErrorHandler has been set.
org.springframework.amqp.rabbit.listener.FatalListenerExecutionException: Illegal null id in message. Failed to manage retry for message: (Body:'{
    <Omitted for privacy>
}'; ID:null; Content:application/json; Headers:{__TypeId__=<OmittedForPrivacy>}; Exchange:; RoutingKey:<OmittedForPrivacy>; Reply:null; DeliveryMode:NON_PERSISTENT; DeliveryTag:1)
    at org.springframework.amqp.rabbit.config.StatefulRetryOperationsInterceptorFactoryBean$3.getKey(StatefulRetryOperationsInterceptorFactoryBean.java:114) ~[spring-rabbit-1.2.1.RELEASE.jar:na]
    at org.springframework.retry.interceptor.StatefulRetryOperationsInterceptor.invoke(StatefulRetryOperationsInterceptor.java:132) ~[spring-retry-1.1.0.RELEASE.jar:na]
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:179) ~[spring-aop-4.0.0.RELEASE.jar:4.0.0.RELEASE]
    at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:207) ~[spring-aop-4.0.0.RELEASE.jar:4.0.0.RELEASE]
    at com.sun.proxy.$Proxy612.invokeListener(Unknown Source) ~[na:na]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.invokeListener(SimpleMessageListenerContainer.java:620) [spring-rabbit-1.2.1.RELEASE.jar:na]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:454) ~[spring-rabbit-1.2.1.RELEASE.jar:na]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:480) [spring-rabbit-1.2.1.RELEASE.jar:na]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:464) [spring-rabbit-1.2.1.RELEASE.jar:na]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$300(SimpleMessageListenerContainer.java:61) [spring-rabbit-1.2.1.RELEASE.jar:na]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:558) [spring-rabbit-1.2.1.RELEASE.jar:na]
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) [na:1.7.0_17]
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) [na:1.7.0_17]
    at java.lang.Thread.run(Thread.java:722) [na:1.7.0_17]

.

Quindi vorrei cambiare il fatto che interrompe il contenitore del ascoltatore dei messaggi e invece mette solo il messaggio sulla coda Deadletter.

KRGDS

Answer 1

Cosa intendi per l'ascoltatore "ferma";Si prega di mostrare la configurazione e pubblica un registro di debug da qualche posto che mostra il comportamento che descrivi.

Modifica:

AH - Il AmqpRejectAndDontRequeueRecover è quello di recuperare dopo le eccezioni dell'ascoltatore;Questa eccezione si verifica prima di entrare anche nella logica dei tentativi.

È possibile aggiungere un MissingMessageIdAdvice alla catena di consulenza (prima del consiglio di riprova).

Lascerà un messaggio con un ID mancante, ma lancerà un RejectAndDontRequeueException se viene ripristinato (e fallirà di nuovo) - solo uno riprova, indipendentemente dall'impostazione del tentativo.

Se vuoi rifiutare tali messaggi senza provare affatto, dovresti creare il tuo consiglio.