introduction à jms 2e partie

Printemps 2004 Annick Fron JMS 2ème partie 1

JMS (2ème partie)

Annick Fron

[email protected]

Http://www.afceurope.com


Rappels du cours précédent

■ JMS (javax.jms) permet de communiquer par messages (non structurés : pas d'interface)

■ 2 modes d'émission : point à point, publish/subscribe■ 2 modes de réception : synchrone (receive bloquant),

asynchrone (MessageListener>>onMessage)■ En PtoP, les messages sont consommés par le

destinataire (<->TCP)■ En publish/subscribe, le subscriber doit être actif au

moment du publish (<->UDP)


Plan : pour une architecture robusteJMS est souvent reconnu comme une architecture de

communication fiable (à la différence du mail) avec une forte qualité de service dont :

■ Le filtrage de messages■ Les priorités, la persistance■ Durable subscribers■ Les accusés de réception : service, applicatifs■ Les transactions (pas de transactions en RMI)


Le filtrage de messages


Filtrage de messages

Filtre

Prop = 'blue'Prop = 'green'

Prop = 'yellow'


Filtrage des messages en PtoPCôté émetteur

message.setStringProperty(" myProp ", " blue ");

Côté récepteurqueueReceiver = session.createReceiver(ioQueue,

"myProp = 'blue'"); // attention aux quotes

Nota : il est impossible de modifier le sélecteur du receiver après création

2 signatures possibles : createReceiver (queue)

createReceiver (queue, selector)

Où selector est une expression logique


Expressions logiques pour le filtrage

Propriétés typées (String, long, boolean, byte, ...)

message.setStringProperty(“propName”,”propValue”);

message.setIntProperty(“size”,10000);

Sous-ensemble de SQL92 : attention aux quotes

(Country = 'UK') OR (Country = 'US') OR (Country = 'France')

age >= 15 AND age <= 19


Filtrage des topics en pubsubOn peut définir des jokers sur les Topics (* pour IBM ou Sonic)

On peut filtrer sur l'arborescence

Exemples :

Tous les résultats

Sport/Football/*/Results

Toutes les équipes

Sport/Football/* (IBM)

Sport.Football.* (Sonic)


Qualité de service


Qualité de service

■ La qualité de service est à la charge du serveur JMS■ Elle varie selon les implémentations :

– IBM Websphere MQ

– Joram

– OpenJMS

– SonicMQ

– SwiftMQ

– Etc…

■ Dans la mesure où elle est implémentée, il est possible de la paramétrer au niveau du code

■ Vérifier les fonctionnalités de l'outil choisi


Structure d'un message

■ Entête– JMSDestination, JMSDeliveryMode, JMSMessageId,

JMSExpiration, JMSPriority, etc.

■ Propriétés– Couple <nom, valeur>

■ Corps– TextMessage, MapMessage

– StreamMessage, ObjectMessage

– BytesMessage

■ Note : le corps n'est pas structuré. Souvent XML


Header d'un message

Champ du header■ JMSDestination■ JMSDeliveryMode■ JMSExpiration■ JMSPriority■ JMSMessageID■ JMSTimestamp■ JMSCorrelationID■ JMSReplyTo■ JMSType■ JMSRedelivered

Défini par ■ send ou publish method■ send ou publish method■ send ou publish method■ send ou publish method■ send ou publish method■ send ou publish method■ Client■ Client■ Client■ JMS provider


Le header d'un message

■ Le header transporte des propriétés de service notamment la priorité, la durée de vie, le mode de livraison (persistant ou non)

■ Tous ces éléments peuvent être décrits :– globalement au niveau du MessageProducer

– de manière intermédiaire au niveau du Message

– de manière locale au moment de l'envoi

■ Toutefois, les outils implémentent l'un ou l'autre niveau mais pas forcément tous

■ Le niveau de l'envoi est le plus fin (send)


Niveaux de qualité de service

MessageProducer

Messagesendsend

sendMessage

send


Envoi d'un message

■ 2 interfaces d'envoi– sender.send(message),

– send(Message message, int deliveryMode, int priority, long timeToLive)

■ Idem publisher.publish(message)■ Idem avec JMS 1.1 unifié : producer.send(...)


Les priorités■ Les priorités vont de 0 (basse) à 9 (haute) : défaut 4■ Les priorités d'un message sont définies dans le

3ème argument lors d'un envoiproducer.send(message,

DeliveryMode.NON_PERSISTENT, 3, 10000);

Nota : l'API mentionne aussi

Au niveau du messagemessage.setJMSPriority(7); // pas dans Sonic

Pour tous les messages d'un producteur (sender/publisher)producer.setPriority(7);// pas dans Sonic


Persistance d'un message■ 2 valeurs possibles :

– DeliveryMode.NON_PERSISTENT (rare)

– DeliveryMode.PERSISTENT (défaut)

■ 2ème argument du send(resp. publish)

publisher.publish(message, DeliveryMode.NON_PERSISTENT, 3, 10000)

■ Idem précédemment pour l'API

■ MessageMessage.setJMSDeliveryMode(DeliveryMode.PERSISTENT)

■ MessageProducer

producer.setDeliveryMode(...)


Nota

■ Le mode de persistance est à la discrétion du serveur de messages.

■ Si le serveur tombe, les messages non persistants sont perdus, les autres sont renvoyés lors de la restauration du serveur

■ Sémantique "one only"

"A message is guaranteed to be delivered once and only once by a JMS provider if the delivery mode of the message is PERSISTENT and if the destination has a sufficient message retention policy."


Définition de la durée de vie d'un message■ Exprimée en ms■ 0 : le message n'expire jamais, 10000 au bout de 10s

(10000 ms) : par défaut n'expire jamais■ 4 ème argument du send

topicPublisher.publish(message, DeliveryMode.NON_PERSISTENT, 3, 10000)

■ Idem pour l'API

■ Aussi sur le producteurproducer.setTimeToLive(10000);

■ Message

Message.setJMSExpiration(10000)


Question

■ A quoi sert de définir la durée de vie d'un message ?


DurableSubscribers


Publish/subscribe

■ En point à point, le message est consommé par le destinataire, dans le délai de son expiration

■ En publish/subscribe, le nombre de destinataires est inconnu : impossible de déterminer qui va louper le message ni combien d'acquittements attendre !

■ Seuls les abonnés actifs à l'envoi du message le reçoivent le message

■ Nota : c'est aussi lié sans doute à UDP


La dure vie des abonnés

M3 - M4 pas reçus !


Les ''durable subscribers''

On n'a manqué aucun épisode !


Principe des durable subscribers

■ Les abonnés durables peuvent recevoir un topic même s'ils n'étaient pas connectés au moment de l'envoi

■ Ceci permet la fiabilité du point à point en mode publish/subscribe

■ Bien sûr au dépend des performances


Scénario

■ Un subscriber s'abonne à un topic■ Lorsque le publisher publie un topic, il le reçoit■ Il ne reçoit plus rien s'il quitte et relance

■ Un Durable subscriber au contraire reçoit tous les messages à la reconnection


Durable subscribers : principe

■ créer une ID client sur la connexion avant de l'ouvrir

conn.setClientID("maConnexion");

conn.start()

■ Envoyer le topic comme persistantproducer.send(mess,DeliveryMode.PERSISTENT,4,30000);

■ créer un abonné durable (durableSubscriber) avec une référence d'abonnement:

session.createDurableSubscriber(topic, "monAbonnement");


Suppression d'un abonnement

■ Fermeture de l'abonné

topicSubscriber.close();

// optionnel en Sonic

■ Suppression de l'abonnement

topicSession.unsubscribe("monAbonnement");


Question

■ On peut préconfigurer la ConnectionFactory avec le client ID.

■ Comment s'assurer que l'application se connecte toujours avec le même client ID ?


Le top de la qualité de service

■ Publish and Subscribe■ Durable Subscriptions■ Persistent Messages (DeliveryMode.PERSISTENT)■ ExceptionListener for Connection Failure■ setPingInterval sur la connection (heartbeat)


Résumé : durable subscribers

■ Permettent la fiabilité du point à point en mode publish/subscribe

■ Nécessitent un ID de connexion, une référence d'abonnement


Accusés de réception


Sémantique one only

Serveur

Retrait 200$

!Conn. timeout

Retrait 200$


En JMS

■ 2 types d'accusés de réception :■ accusé de "service" indiquant à l'infrastructure que le

message peut être supprimé ; il n'est pas vu de l'application. Ils sont liés à la durée de vie du message, à son mode de livraison (persistant ou non), et au mode d'envoi (point à point, pub/sub)

■ accusé de réception applicatif en direction de l'émetteur, programmé à la main


Rappel : mode point à point

Il ne peut y avoir qu'un seul consommateur.Le message est ensuite retiré après acquittement


Rappel : mode pub/sub

Pour les durable subscribers comme point à point , pour les autres subscribers message retiré en fin de session


Acknowledge

■ Pour savoir quand le message a été consommé, on utilise un acknowledge : implicite ou par programme

■ Le type d'acknowledge est associé à la Session lors de sa création

queueConnection.createQueueSession(

transacted,

Session.AUTO_ACKNOWLEDGE)


Question

■ Question : pourquoi la session, pas le message ?


Rappels Session

■ Une Session peut gérer plusieurs émetteurs/récepteurs, envois/réceptions : scénario

sess

ion

receive

send

receive

send


Note : connexions et sessions

Session

Session

Session

Connexion

Client

Il peut y avoir plusieurssession dans une connexionUne connexion est liée àun serveur (ex : maui)


Acknowledge


Accusés de réception de service

■ Plusieurs types d'accusés de réception :

■ Session.AUTO_ACKNOWLEDGE : accusé automatique

■ Session.CLIENT_ACKNOWLEDGE : accusé explicite■ Session.DUPS_OK_ACKNOWLEDGE : accusé

retardé, plusieurs accusés possibles


Différents types d'acknowledge

■ Session.AUTO_ACKNOWLEDGE :– Envoyé dès que le récepteur termine onMessage en

asynchrone, receive en synchrone

■ Session.CLIENT_ACKNOWLEDGE– L'acknowledge est global au niveau de la session (pour tous

les messages jusqu'à présent) et déclenché explicitement côté réception par :

message.acknowledge()


Différents types (suite)

Session.DUPS_OK_ACKNOWLEDGE■ Accusé de réception "paresseux"■ A n'utiliser que par les récepteurs qui peuvent

accepter des messages dupliqués■ Minimise le travail de la session pour éviter les

doublons■ => plus de sémantique "one only"


Modes d'accusés de réceptionNouveaumessage !

Tu accapares la bande

passante !

CLIENT

J'ai reçu 50 messages

jusqu'à présent

Paresseux si tu veux, mais n'importe

si je reçois 2 fois

Que fais-tu ? Je viens d'envoyer 50

messages

Tu m'obliges à envoyerles messages plusieurs

fois !

AUTO

DUPS_OK


Régles de retrait

■ Quand une QueueSession (ou une TopicSession avec DurableSubscribers) termine sans acknowledgement– Le serveur garde le message pour le relivrer lors de

l'ouverture de la session suivante

■ Les messages sans accusé d'un non-durable subscriber (en mode pub/sub) sont retirés lors de la fermeture de sa session


Session sans ack


Non ack recovery


Mode non persistant : ack OK


Mode non persistant : pas d'ack


Accusés de réception : résumé

■ Défini au niveau de la session createQueueSession(boolean transacted,

int acknowledgeMode)

3 types :■ AUTO_ACKNOWLEDGE■ CLIENT_ACKNOWLEDGE■ DUPS_OK_ACKNOWLEDGE■ Si l'acknowledge est explicite, le client doit écrire :

message.acknowledge();


Accusés de réception applicatifs


Accusés de réception applicatifs

Emet/récept A Emet/récept B

1

2

3

4

5

Est-ce que 3 est la réponse à 1 ? À 2 ?


Destinations temporaires

■ On peut utiliser des TemporaryQueues pour faire le lien : analogie enveloppe dans une enveloppe

■ Côté envoyeurmessage.setJMSReplyTo(session.createTemporaryQueue())

■ Côté clientproducer = session.createProducer(msg.getJMSReplyTo());

replyMsg = session.createTextMessage("Consumer " +

"processed message: " + msg.getText());

replyMsg.setJMSCorrelationID(msg.getJMSMessageID());

producer.send(replyMsg);

Les destinations temporaires n'existent que le temps de

la connexion qui les a créées


Destination pour la réponse

émetteur

récepteur


QueueRequestor

■ Il existe un framework plus systématique : QueueRequestor

Le QueueRequestor utilise 2 principes :■ la destination de réponse (TemporaryQueue définie

dans getJMSReplyTo())■ une corrélation d'ID (rappelez-vous du header !)

getJMSCorrelationID()


QueueRequestor : côté envoyeur

■ queueRequestor = new QueueRequestor(queueSession, queue)

■ Il n'y a pas d'envoi explicite du message, mais au contraire, attente d'une réponse :

reply = (TextMessage) queueRequestor.request(message)

■ Ceci remplace :

sender.send(message)


QueueRequestor : côté récepteur

■ On reçoit le message normalement : queueReceiver.receive()

■ On extrait du message une queue temporaire définie implicitement par le requestor (enveloppe de retour) :

tempQueue = (Queue) message.getJMSReplyTo();

■ On envoie un message classiquement dans cette Queue en prenant soin de fournir l'ID de corrélation :

replySender = queueSession.createSender(tempQueue);

reply = queueSession.createTextMessage(“texte mess”);

reply.setJMSCorrelationID(message.getJMSMessageID());

replySender.send(reply);


QueueReceptor(3) : retour à l'envoyeur■ On vérifie l'ID de corrélation du message de départ

replyID = new String(reply.getJMSCorrelationID());

messageID = message.getJMSMessageID()

replyID.equals(messageID) // oui


Résumé : accusés de réception applicatifs■ Il est possible d'implémenter des accusés de

réception applicatifs en utilisant des destinations temporaires

■ Le header du message transporte MessageId et MessageCorrelationID

■ Le mécanisme de QueueRequestor offre un framework plus complet et permet de s'assurer de la corrélation de deux messages en sens inverses


Les transactions


Les transactions : principe

■ Une session peut être transactionnelle■ Côté émetteur :

– Les messages envoyés ne sortent pas tant avant "commit"

– Les messages envoyés sont retirés si "rollback"

■ Côté récepteur– Les messages reçus ne sont pas acquittés avant "commit"

– Les messages reçus sont relivrés lors de "rollback"


Les transactions

■ La connection crée d'abord une (Q/T)session avec transaction (transacted = true):

connexion.createQueueSession(true, Session.AUTO_ACKNOWLEDGE)

■ On peut alors utiliser :

queueSession.commit();

queueSession.rollback();


Transaction

■ Les transactions sont consécutives

transaction1 transaction2

Commit/rollback

Il est très difficile de gérer des transactions gigognes(nested transactions)


Transactions locales


Transactions (suite)

2 clients ne peuvent partager la même transaction. Par contre un producteur peut envoyerun message à plusieurs récepteurs dans la même transaction


Transactions (suite)

■ On appelle toujours le rollback sur exception

catch (Exception e) {session.rollback(); …}


Exemple typique : cohérence d'un workflow

Service achatscommande livraison

transaction

receive send

session.createReceiver(« commande »);receiver.receive();session.createSender(« livraison »)sender.send(...)session.commit()


2 phases commit

■ Le 2-phase commit est une généralisation sur plusieurs transactions : pas de commit global tant que pas tous les commits individuels

■ Note : pour le two-phase commit, il est possible de compléter avec l'interface JTA/XA, mais Sun recommande d'utiliser les services propriétaires de l'implémenteur.

■ XAQueueConnectionFactory

eStore BDpaiement stockage livraison


Transactions : résumé

■ Une session JMS peut être transactionnelle

connexion.createQueueSession(true, 0)

■ commit et rollback concernent l'ensemble des messages de la session

■ Souvent utilisé en workflow■ Il existe des extensions JTA/XA

(XAQueueConnexionFactory) pour le 2-phase commit (Java Transaction API) y compris avec des BD


Conclusions

■ Le point fort de JMS est la sémantique « one and only one » ainsi que la qualité de service (Qos)

■ On peut définir des priorités de messages, des durées de vie de messages, la persistance

■ Les accusés de réception de service déterminent la rétention des messages

■ Il existe un framework applicatif d'accusé de réception■ JMS supporte les transactions (-> EJB et Message-

Driven Beans)


Références

■ http://www.hartmannsoftware.com/JMS_Tutorial.pdf■ http://www-106.ibm.com/developerworks/java/library/j-jmsvendor/?dwzone=java■ http://java.sun.com/products/jms/tutorial/

introduction à jms 2e partie

Documents