Nell’articolo Code JMS (parte 1) abbiamo parlato di code JMS, di come si configurano sull’Application Server WildFly e di come sia possibile inviare e leggere messaggi da una semplice coda. In questo articolo implementiamo un listener per leggere nella coda e parliamo del meccanismo publish e subscribe.
Listener
Nell’articolo Primi Passi con gli EJB3 abbiamo brevemente accennato al fatto che i Message Drive Bean sono utilizzati per leggere i messaggi da una coda JMS in modo asincrono. In molti aspetti un MDB è simile ad uno Stateless Session Bean, infatti essi non hanno uno stato associato ad uno specifico client. Per questo il container può gestirli in un pool da cui li estrae per processare un nuovo messaggio nella coda.
La scrittura di un MDB richiede l’implementazione dell’interfaccia MessageListener che definisce il solo metodo onMessage() invocato dal container per il processamento del messaggio prelevato dalla coda. Con riferimento al codice sorgente presentato nell’articolo citato nell’introduzione, l’MDB che legge dalla coda JBQ si presenta nel seguente modo.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 |
@MessageDriven(activationConfig = { @ActivationConfigProperty(propertyName = "destinationLookup", propertyValue = "java:/jms/queue/javaboss"), @ActivationConfigProperty(propertyName = "destinationType", propertyValue = "javax.jms.Queue") }) public class JBQListener implements MessageListener { public void onMessage(Message message) { try { System.out.println( "MDB: " + message.getBody( String.class ) ); } catch (JMSException e) { e.printStackTrace(); } } } |
L’annotazione @MessageDriven identifica l’oggetto come un MDB e ne specifica la configurazione attraverso la proprietà activationConfig e l’annotazione @ActivationConfigProperty. Questa è caratterizzata da coppie chiave-valore che sono utilizzate dal provider per configurare l’MDB. In particolare nel nostro esempio sono valorizzate le seguenti proprietà (minime):
destinationLookup: la coda o topic di ascolto;destinationType: il tipo di oggetto atteso trajavax.jms.Queueejavax.jms.Topic.
Una volta avviato WildFly noterete nel log la seguente stringa che indica la registrazione dell’MDB JBQListener.
|
1 |
JBAS014142: Started message driven bean 'JBQListener' with 'hornetq-ra.rar' resource adapter |
Topic
Negli esempi abbiamo mostrato come inviare e leggere i contenuti da una coda. Vediamo ora come utilizzare un topic e come sia possibile leggere un messaggio attraverso in MDB filtrando solamente alcuni messaggi di interesse.
Vediamo innanzitutto la classe JBTWriter per la scrittura del messaggio nella coda che si presenta nel modo seguente:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 |
@Singleton public class JBTWriter { @Resource(name = "java:/jms/topic/javaboss") private Topic jbt; @Resource(mappedName = "java:/JmsXA") private ConnectionFactory cf; private Connection connection; public void sendMessage(String txt, Integer type) { try { connection = cf.createConnection(); Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); MessageProducer publisher = session.createProducer(jbt); connection.start(); TextMessage message = session.createTextMessage(txt); message.setIntProperty( "type", type ); publisher.send(message); } catch (Exception exc) { exc.printStackTrace(); } finally { if (connection != null) { try { connection.close(); } catch (JMSException e) { e.printStackTrace(); } } } } } |
La principale differenza con il bean JBQWriter è che attraverso l’annotation @Resource è iniettato un oggetto javax.jms.Topic che identifica la coda.
Inoltre l’oggetto MessageText eredita da javax.jms.Message un set di metodi che consentono di associare al messaggio parametri aggiuntivi di diverso tipo, che saranno trasportati nel payload del messaggio. Nel nostro esempio al messaggio è legato un parametro type di tipo intero.
Il Message Driven Bean che legge i messaggi dal topic utilizzerà il parametro per filtrare solamente i messaggi specifici. Per farlo è necessario valorizzare la proprietà messageSelector con l’espressione di selezione come mostrato nel codice seguente.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 |
@MessageDriven(activationConfig = { @ActivationConfigProperty(propertyName = "destinationLookup", propertyValue = "java:/jms/topic/javaboss"), @ActivationConfigProperty(propertyName = "destinationType", propertyValue = "javax.jms.Topic"), @ActivationConfigProperty(propertyName = "messageSelector",propertyValue = "type = 1") }) public class JBTListenerType1 implements MessageListener { public void onMessage(Message message) { try { System.out.println( "Type 1 MDB: " + message.getBody( String.class ) ); } catch (JMSException e) { e.printStackTrace(); } } } |
La sintassi dell’espressione è basata su un sottoinsieme della specifica SLQ92.
Nel progetto è presente anche una MDB JBTListenerType2 che esegue la selezione con l’espressione "type = 2".
Modalità di Acknowledge
I messaggi in coda non vengono rimossi fino a quando non è invia l’ACK. L’elaborazione con successo di un messaggio in genere avviene in tre step:
- il receiver riceve il messaggio;
- il receiver processa il messaggio;
- l’ACK è inviato.
L’invio dell’ACK può avvenire da parte del provider o del consumer in base alla modalità di acknowledge. In una sessione transazionale l’ACK è inviato in modo automatico al commit della transazione. Diversamente in caso di rollback tutti i messaggi consumati vengono reinseriti in coda (e redelivered).
In sessioni non transazionali quando e come è inviato l’ACK dipende da come è configurata la sessione. Tale configurazione è specificata nell’invocazione del metodo createSession(), che la prevede come secondo parametro (vedi esempi riportati), o, nel caso di MDB, utilizzando @ActivationConfigProperty nel modo seguente:
|
1 |
@ActivationConfigProperty(propertyName="acknowledgeMode",propertyValue="...") |
dove propertyValue specifica la modalità da utilizzare. I valori ammessi sono.
| AUTO_ACKNOWLEDGE | La sessione invia automaticamente l’ACK nel momento in cui il receiver esegue con successo il metodo receive() o quando l’MDB ritorna con successo dal metodo onMessage(): |
| CLIENT_ACKNOWLEDGE | Il receiver invia automaticamente l’ACK invocando il metodo acknowledge sull’oggetto Message. L’ACK avviene a livello di sessione, ovvero tutti i messaggi letti nella sessione ricevono l’ACK anche se è invocato solamente sull’ultimo letto. |
| DUPS_OK_ACKNOWLEDGE | In questa modalità la sessione invia l’ACK quando meglio crede (lazy). Questa modalità potrebbe generare la consegna di messaggi già elaborati, quindi il receiver deve essere in grado di riconoscerli ed elaborarli correttamente. |
Se i messaggi sono consumati dalla coda ma non hanno ricevuto l’ACK al termine della sessione, il provider li conserva e li riconsegna non appena un client si connette alla coda.
Durability
Nel post Code JMS (parte 1) abbiamo detto che per ricevere un messaggio da un topic è necessaria una sottoscrizione che deve essere attiva al momento dell’invio del messaggio da parte del client. Questo richiede al consumer di essere sempre connesso al topic, altrimenti non riceverà i messaggi inseriti nella coda dopo la sua sconnessione e fino alla sua riconnessione.
L’eccezione a tale comportamento è conosciuta come durable subscription: una volta che il consumer ottiene una durable subscription al topic il provider si assicura che tutti i messaggi gli siano inviati, anche se il consumer non è connesso. Se l’MDB desidera instaurare una durable subscriber allora l’ActivationConfigProperty dovrebbe apparire nel seguente modo:
|
1 |
@ActivationConfigProperty(propertyName="subscriptionDurability", propertyValue="Durable") |
Codice Sorgente
Il codice sorgente del progetto presentato è disponibile qui jms-2.