Consumer empfängt keine Nachrichten von ActiveMQ

Question

Wir stehen vor einem zufälligen Problem mit ActiveMQ und seinen Kunden. Wir beobachten, dass wenige Verbraucher keine Nachrichten empfangen, obwohl sie mit der ActiveMQ-Warteschlange verbunden sind. Aber es funktioniert gut nach dem Neustart des Verbrauchers.

Wir haben eine Warteschlange mit dem Namen testQueue auf ActiveMQ-Seite. Ein Benutzer versucht, die Nachrichten aus dieser Warteschlange zu entfernen. Zu diesem Zweck verwenden wir den DefaultMessageListenerContainer von Spring. Die Nachricht wird von ActiveMQ Broker an den Consumerknoten gesendet. Auch vom tcpdump war es offensichtlich, dass die Nachricht den Consumer-Knoten erreicht, aber der eigentliche Consumer-Code die Nachricht nicht sehen kann. Mit anderen Worten, die Nachricht scheint entweder im ActiveMQ-Consumer-Code oder in Spring's DefaultMessageListenerContainer hängen zu bleiben.

Siehe die folgende Abbildung. für mehr Klarheit in der Sache. Die Nachricht erreicht den Consumer-Knoten, aber sie erreicht nicht die "tatsächliche Consumer-Klasse", was bedeutet, dass die Nachricht entweder im AMQ-Consumer-Code oder in Spring DMLC hängen geblieben ist.

Im Folgenden sind die von ActiveMQ Admin erfasst Details.

Queue-Name/Pending-Nachricht-Count/Consumer-Count/Nachrichten der Warteschlange gestellte/Nachrichten-Aus der Warteschleife angenommene heißt test/9/1/9/0

Im Folgenden sind die Details.

Anschluss-ID/SessionId/Wählhebel/Enqueues/reiht/Versand/Versand-Queue/Prefetch ID: bearsvir52-45176-1375519181268-3: 5/1// 9/0/9/9/250

Aus der zweiten Tabelle ist es offensichtlich, dass Nachrichten an den Verbraucher geliefert werden, aber der Verbraucher die Nachricht nicht anerkennt. Daher sind die Nachrichten auf der Brokerseite in der Dispatched-Warteschlange hängen geblieben.

Nur wenige Punkte zu Ihrer Mitteilung:

1) Es gibt keine Zeitdifferenz s/w-Broker-Knoten und Verbraucherknoten.

2) Beobachtete den tcpdump auf der Verbraucherseite. Wir können MessageDispatch (Openwire) -Paket sehen, das zum Verbraucherknoten übertragen wird, aber das MessageAck (Openwire) für das selbe nicht finden konnte.

3) Manchmal funktioniert es auf einem Knoten, und manchmal erzeugt es ein Problem auf demselben Knoten.

Answer 1

Es dauerte viel Zeit, um die Lösung zu finden. Es scheint ein Problem mit der Klasse org.apache.activemq.ActiveMQConnection.java zu geben, falls AMQ ausfällt. Das Verbindungsobjekt wird in solchen Fällen nicht auf Verbraucherseite gestartet.

Im Folgenden ist der Fix Ich habe in ActiveMQConnection.java Datei hinzugefügt und kompilierte die Quellen zu erstellen activemq-core-x.x.x.jar

private final Object startMutex = new Object(); 

einen Scheck in create Methode

public Session createSession(boolean transacted, int acknowledgeMode) throws JMSException { 
    synchronized (startMutex) { 
     if(!isStarted()) { 
      start(); 
     } 
    } 

Answer 2

Eine Ursache dafür kann die falsche Verwendung eines CachingConnectionFactory (mit zwischengespeicherten Konsumenten) mit einem Listener-Container sein, der die Konsumenten dynamisch anpasst (max consumenten> consumers). Sie können mit einem zwischengespeicherten Verbraucher enden, der gerade im Pool sitzt und nicht aktiv genutzt wird. Sie müssen Benutzer niemals mit einem Listener-Container zwischenspeichern.

Für solche Probleme empfehle ich im Allgemeinen mit TRACE-Protokollierung zu laufen und Sie können alle Verbraucheraktivitäten sehen.