Wie pausiere und setze ich den asynchronen Verbrauch von JMS-Nachrichten fort

Question

Ich bin eine Anwendung mit ActiveMQ erstellen, wo ich einen Produzenten und einen Verbraucher habe. In der Consumer Iam mit einem MessageListener, um Nachrichten von Producer asynchron abhören, die mit einer Methode namens OnMessage (Message-Nachricht) erfolgt ist. Aber bevor ich die Nachrichten konsumiere, möchte ich eine Zustandsprüfung durchführen und dann die Nachrichten konsumieren. Ich möchte keinen synchronen Nachrichtenverbrauch verwenden, da dies gegen mein Design verstößt.

void initialize() throws JMSException { 
      this.connection = this.connectionFactory.createConnection(); 
      this.connection.start(); 
      final Session session = this.connection.createSession(false, Session.AUTO_ACKNOWLEDGE); 
      final Destination destination = session.createQueue("testQ"); 
      this.consumer = session.createConsumer(destination); 
      this.consumer.setMessageListener(this); 
} 

prüfen Zustand hier wie Erkennung von Internet-Verbindung usw.

public void onMessage(final Message message) { 
     Preconditions.checkNotNull(message); 

     if (!(message instanceof TextMessage)) { 
      _LOG.error("The message is not of type TextMessage but of type {} so we could not process", message.getClass().getSimpleName()); 
      throw new IllegalArgumentException("This type '" + message.getClass().getSimpleName() + "' of message could not been handled"); 
     } 

     try { 
      final String messageType = message.getStringProperty("messageType"); 
      Preconditions.checkNotNull(messageType); 
      _LOG.info("The MessageType is {}", messageType); 

      final String msg = ((TextMessage) message).getText(); 
      Preconditions.checkNotNull(msg); 
      _LOG.debug(msg); 

      process(messageType, msg); 
     } catch (final JMSException e) { 
      _LOG.error("We could not read the message", e); 
     } 
    } 

Jeder Code Beispiel wäre toll.

Answer 1

Es ist nicht klar, warum Sie Prüfungen durchführen möchten, denn wenn es ein Problem mit der Verbindung gibt, wird Ihre Anwendung über diesen Fehler informiert. Sie können einen ExceptionListener für JMS Connection einrichten, der aufgerufen wird, wenn bei der Verbindung ein Problem auftritt.

Die onMessage wird nicht aufgerufen, wenn ein Verbindungsproblem vorliegt.

Auch ich würde die connection.start() -Methode Anruf nach dem Einrichten der Nachricht Listener für Verbraucher schieben. Ein Aufruf von connection.start() zeigt dem Messaging-Provider an, dass die Anwendung zum Empfang von Nachrichten bereit ist.

Es gibt connection.stop(), um die Nachrichtenzustellung anzuhalten/zu stoppen. Sie können connection.start() erneut ausgeben, um die Nachrichtenzustellung fortzusetzen.

---

-Update auf der Grundlage Ihrer Antwort

Ich schlage vor, Sie eine Sitzung mit Client-Modus verwenden, bestätigen.

final Session session = this.connection.createSession(false, Session.CLIENT_ACKNOWLEDGE) 

dann in onMessage Methode, wenn es keine Internet-Verbindung ist, bestätigt die Nachricht nicht. Die Nachricht wird erneut gesendet, wenn sie nicht abgelaufen ist.

Answer 2

In der asynchronen Nachrichtenübermittlung können Sie die Bedingung vor der Verarbeitung der Nachricht immer überprüfen. Sobald die JMS-Nachricht an Ihren Listener-Code (d. H. OnMessage-Methode) empfangen wurde, liegt es an Ihnen, die weitere Richtung anzugeben. Ich hatte ein Problem gelöst, bei dem ich die JMS-Nachricht mit meinem db-Inhalt überprüfen wollte.

Kontrolle der Internet-Verbindung am Leben ist von URLConnection an den Webservice-URL so etwas wie unten machen:

public static boolean isReachable(String targetUrl) throws IOException 
{ 
    try { 
    HttpURLConnection httpUrlConnection = (HttpURLConnection) new URL(
     targetUrl).openConnection(); 
    httpUrlConnection.setRequestMethod("HEAD"); 


    int responseCode = httpUrlConnection.getResponseCode(); 

    return responseCode == HttpURLConnection.HTTP_OK; 
} catch (UnknownHostException noInternetConnection) 
    { 
    return false; 
    } 
} 

Und dann in Ihrem onMessage Methode rufen Sie die Methode als

public void onMessage(final Message message) { 
    if(isReachable){ 
    Preconditions.checkNotNull(message); 

    if (!(message instanceof TextMessage)) { 
     _LOG.error("The message is not of type TextMessage but of type {} so we could not process", message.getClass().getSimpleName()); 
     throw new IllegalArgumentException("This type '" + message.getClass().getSimpleName() + "' of message could not been handled"); 
    } 

    try { 
     final String messageType = message.getStringProperty("messageType"); 
     Preconditions.checkNotNull(messageType); 
     _LOG.info("The MessageType is {}", messageType); 

     final String msg = ((TextMessage) message).getText(); 
     Preconditions.checkNotNull(msg); 
     _LOG.debug(msg); 

     process(messageType, msg); 
    } catch (final JMSException e) { 
     _LOG.error("We could not read the message", e); 
    } 
    } 
}