Il consumatore non riceve messaggi da ActiveMQ

Question

Siamo di fronte a un problema casuale con ActiveMQ e i suoi utenti. Osserviamo che pochi consumatori non ricevono messaggi, anche se sono collegati alla coda ActiveMQ. Ma funziona bene dopo il riavvio del consumatore.

Abbiamo una coda denominata testQueue sul lato ActiveMQ. Un consumatore sta tentando di disattivare i messaggi in coda dalla coda. A tale scopo utilizziamo Spring's DefaultMessageListenerContainer. Il messaggio viene consegnato al nodo del consumatore da ActiveMQ Broker. Dal tcpdump, era ovvio che il messaggio stava raggiungendo il nodo del consumatore, ma il codice del consumatore effettivo non è in grado di vedere il messaggio. In altre parole, il messaggio sembra essere bloccato nel codice del consumatore ActiveMQ o in DefaultMessageListenerContainer di Spring.

Vedere riferimento alla figura seguente. per maggiore chiarezza sulla questione. Il messaggio sta raggiungendo il nodo Consumer, ma non sta raggiungendo la "Classe di consumo effettivo", il che significa che il messaggio è rimasto bloccato nel codice del consumatore AMQ o nel DMLC Spring.

Sotto i dettagli catturati da ActiveMQ admin.

Queue-Nome/In attesa-Message-Count/Consumer-Count/Messaggi-accodato/Messaggi-rimosse dalla coda test/9/1/9/0

seguito sono elencate le più particolari.

connessione-ID/SessionId/selettore/accoda/Ritiri dalla coda/Inviato/Inviato-coda/Prefetch ID: bearsvir52-45176-1375519181268-3: 5/1// 9/0/9/9/250

Dalla seconda tabella è ovvio che i messaggi vengono recapitati al consumatore, ma il consumatore non sta riconoscendo il messaggio. Quindi i messaggi sono bloccati in Dispatch-Queue sul lato broker.

pochi punti per alla vostra attenzione:

1) Non v'è differenza di tempo b/w nodo Broker e il nodo dei consumatori.

2) Osservato il tcpdump dal lato del consumatore. Possiamo vedere il pacchetto MessageDispatch (Openwire) trasferito al nodo del consumatore, ma non è stato possibile trovare il MessageAck (Openwire) per lo stesso.

3) A volte funziona su un nodo e talvolta crea problemi sullo stesso nodo.

Answer 1

Ci vuole molto tempo per capire la soluzione. Sembra esserci qualche problema con la classe org.apache.activemq.ActiveMQConnection.java, in caso di failover AMQ. In questi casi, l'oggetto di connessione non viene avviato dal lato del consumatore.

Di seguito è riportata la correzione che ho aggiunto nel file ActiveMQConnection.java e ho compilato le origini per creare activemq-core-x.x.x.JAR

private final Object startMutex = new Object(); 

aggiunto un controllo di metodo createSession

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

Answer 2

Una causa di ciò può essere erroneamente utilizzare un CachingConnectionFactory (con utenti memorizzati nella cache) con un contenitore listener che aggiusta dinamicamente i consumatori (utenti massimi> consumatori). È possibile ritrovarsi con un consumatore memorizzato nella cache che si trova nella piscina e non viene utilizzato attivamente. Non è mai necessario memorizzare nella cache i consumatori con un contenitore listener.

Per problemi di questo tipo, in genere consiglio di eseguire il log di TRACE e di visualizzare tutte le attività del consumatore.