Wie man Nachrichten zerstörend aus RabbitMq-Warteschlange mit SimpleMessageListenerContainer Bean

Question

Ich brauche eine Hilfe, um einen SimpleMessageListenerContainer in einer Weise zu implementieren, dass es die Nachricht untransaktional entfernt, sobald sie vom Nachrichtenempfänger gelesen wird.

In meinem Fall, unabhängig von erfolgreichen oder nicht erfolgreichen Transaktionen, die Nachricht in der Warteschlange hängen irgendwo (nicht in der Warteschlange), und wird wiederholt bei jeder Operation, die aus der Warteschlange liest verarbeitet. Daher bleiben alle anderen Nachrichten, die in die Warteschlange eingereiht werden, nicht zugänglich, und nur die erste Nachricht wird jedes Mal erneut verarbeitet. Das andere seltsame Ding ist, dass ich die Nachrichten, die in der Warteschlange landet/Schlange stehen, nicht sehen kann, dh die Warteschlangentiefe ändert sich nie auf der Rabbit Management Konsole, sondern nur die Nachrichtenrate nimmt einen Sprung bei jedem Schreiben in die Warteschlange .

Unten ist das Code-Snippet meiner Java-Konfiguration. Es wird eine große Hilfe sein, wenn jemand hier den Fehler hinweisen können: -

@Configuration 
@EnableJpaRepositories(basePackages={"com.xxx.muppets"}) 
public class MuppetMessageConsumerConfig { 

private ApplicationContext applicationContext; 
@Value("${rabbit.queue.name}") 
private String queueName; 

@Autowired 
public void setApplicationContext(ApplicationContext applicationContext) { 
    this.applicationContext = applicationContext; 
} 

@Bean 
Queue queue() { 
    return new Queue(queueName, false); 
} 

@Bean 
TopicExchange exchange() { 
    return new TopicExchange("spring-boot-exchange"); 
} 

@Bean 
Binding binding(Queue queue, TopicExchange exchange) { 
    return BindingBuilder.bind(queue).to(exchange).with(queueName); 
} 

@Bean 
MuppetMsgReceiver muppetMsgReceiver(){ 
    return new MuppetMsgReceiver(); 
} 

@Bean 
MessageListenerAdapter listenerAdapter(MuppetMsgReceiver muppetMsgReceiver){ 
    return new MessageListenerAdapter(muppetMsgReceiver, "receiveMessage"); 
} 

@Bean 
SimpleMessageListenerContainer container(ConnectionFactory connectionFactory, MessageListenerAdapter listenerAdapter) { 
    SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(); 
    container.setAcknowledgeMode(NONE); 
    container.setConnectionFactory(connectionFactory); 
    container.setQueueNames(queueName); 
    container.setMessageListener(listenerAdapter); 
    return container; 
} 
} 

Meine Nachricht Empfänger Klasse ist wie folgt:

public class MuppetMsgReceiver { 

private String muppetMessage; 

private CountDownLatch countDownLatch; 

public MuppetMsgReceiver() { 
    this.countDownLatch = new CountDownLatch(1); 
} 

public MuppetMsgReceiver(CountDownLatch latch) { 
    this.countDownLatch = latch;   
    CountDownLatch getCountDownLatch() { 
    return countDownLatch; 
} 

public void receiveMessage(String receivedMessage) { 
    countDownLatch.countDown(); 
    this.muppetMessage = receivedMessage; 
} 

public String getMuppetMessage() { 
    return muppetMessage; 
} 
} 

Dieser vollständige Code auf dem getting started example des Frühlings basiert aber hilft nicht durch zerstörungsfreie Lesevorgänge aus der Warteschlange.

Answer 1

AcknowledgeMode=NONE bedeutet den Broker acks die Nachricht, sobald es an den Kunden versandt, nicht, wenn der Verbraucher erhält es so Ihr beobachtete Verhalten für mich keinen Sinn macht. Es ist ziemlich ungewöhnlich, diesen Ack-Modus zu verwenden, aber die Nachricht wird gelöscht, sobald sie gesendet wird. Dies ist wahrscheinlich ein Problem mit Ihrem Code. Mit der Protokollierung von TRACE können Sie sehen, was passiert.