Filtro del ricevitore di messaggi JMS da JMSCorrelationID
Domanda
Come posso creare un'istanza di un listener di code JMS in java (JRE / JDK / J2EE 1.4) che riceve solo messaggi che corrispondono a un determinato JMSCorrelationID? I messaggi che sto cercando di raccogliere sono stati pubblicati in una coda e non in un argomento, anche se questo può cambiare se necessario.
Ecco il codice che sto attualmente utilizzando per mettere il messaggio in coda:
/**
* publishResponseToQueue publishes Requests to the Queue.
*
* @param jmsQueueFactory -Name of the queue-connection-factory
* @param jmsQueue -The queue name for the request
* @param response -A response object that needs to be published
*
* @throws ServiceLocatorException -An exception if a request message
* could not be published to the Topic
*/
private void publishResponseToQueue( String jmsQueueFactory,
String jmsQueue,
Response response )
throws ServiceLocatorException {
if ( logger.isInfoEnabled() ) {
logger.info( "Begin publishRequestToQueue: " +
jmsQueueFactory + "," + jmsQueue + "," + response );
}
logger.assertLog( jmsQueue != null && !jmsQueue.equals(""),
"jmsQueue cannot be null" );
logger.assertLog( jmsQueueFactory != null && !jmsQueueFactory.equals(""),
"jmsQueueFactory cannot be null" );
logger.assertLog( response != null, "Request cannot be null" );
try {
Queue queue = (Queue)_context.lookup( jmsQueue );
QueueConnectionFactory factory = (QueueConnectionFactory)
_context.lookup( jmsQueueFactory );
QueueConnection connection = factory.createQueueConnection();
connection.start();
QueueSession session = connection.createQueueSession( false,
QueueSession.AUTO_ACKNOWLEDGE );
ObjectMessage objectMessage = session.createObjectMessage();
objectMessage.setJMSCorrelationID(response.getID());
objectMessage.setObject( response );
session.createSender( queue ).send( objectMessage );
session.close();
connection.close();
} catch ( Exception e ) {
//XC3.2 Added/Modified BEGIN
logger.error( "ServiceLocator.publishResponseToQueue - Could not publish the " +
"Response to the Queue - " + e.getMessage() );
throw new ServiceLocatorException( "ServiceLocator.publishResponseToQueue " +
"- Could not publish the " +
"Response to the Queue - " + e.getMessage() );
//XC3.2 Added/Modified END
}
if ( logger.isInfoEnabled() ) {
logger.info( "End publishResponseToQueue: " +
jmsQueueFactory + "," + jmsQueue + response );
}
} // end of publishResponseToQueue method
Soluzione
L'impostazione della connessione della coda è la stessa, ma una volta che hai QueueSession, imposti il ??selettore quando crei un ricevitore.
QueueReceiver receiver = session.createReceiver(myQueue, "JMSCorrelationID='theid'");
poi
receiver.receive()
o
receiver.setListener(myListener);
Altri suggerimenti
A proposito, sebbene non sia la vera domanda che hai posto - se stai cercando di implementare la risposta alla richiesta su JMS, ti consiglio leggere questo articolo poiché l'API JMS è un po 'più complessa di quanto si possa immaginare e farlo in modo efficiente è molto più difficile di quanto sembri.
In particolare per utilizzare JMS in modo efficiente dovresti cerca di evitare di creare consumatori per un singolo messaggio ecc.
Anche perché l'API JMS è così complessa da usare correttamente ed efficientemente - in particolare con pooling, transazioni ed elaborazione simultanea - raccomando a persone nasconde il middleware dal codice dell'applicazione , ad esempio tramite l'utilizzo di Implementazione di Spring Remoting di Apache Camel per JMS
Spero che questo possa aiutare. Ho usato Open MQ.
package com.MQueues;
import java.util.UUID;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.QueueConnection;
import javax.jms.QueueReceiver;
import javax.jms.QueueSession;
import javax.jms.Session;
import javax.jms.TextMessage;
import com.sun.messaging.BasicQueue;
import com.sun.messaging.QueueConnectionFactory;
public class HelloProducerConsumer {
public static String queueName = "queue0";
public static String correlationId;
public static String getCorrelationId() {
return correlationId;
}
public static void setCorrelationId(String correlationId) {
HelloProducerConsumer.correlationId = correlationId;
}
public static String getQueueName() {
return queueName;
}
public static void sendMessage(String threadName) {
correlationId = UUID.randomUUID().toString();
try {
// Start connection
QueueConnectionFactory cf = new QueueConnectionFactory();
QueueConnection connection = cf.createQueueConnection();
QueueSession session = connection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
BasicQueue destination = (BasicQueue) session.createQueue(threadName);
MessageProducer producer = session.createProducer(destination);
connection.start();
// create message to send
TextMessage message = session.createTextMessage();
message.setJMSCorrelationID(correlationId);
message.setText(threadName + "(" + System.currentTimeMillis()
+ ") " + correlationId +" from Producer");
System.out.println(correlationId +" Send from Producer");
producer.send(message);
// close everything
producer.close();
session.close();
connection.close();
} catch (JMSException ex) {
System.out.println("Error = " + ex.getMessage());
}
}
public static void receivemessage(final String correlationId) {
try {
QueueConnectionFactory cf = new QueueConnectionFactory();
QueueConnection connection = cf.createQueueConnection();
QueueSession session = connection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
BasicQueue destination = (BasicQueue) session.createQueue(getQueueName());
connection.start();
System.out.println("\n");
System.out.println("Start listen " + getQueueName() + " " + correlationId +" Queue from receivemessage");
long now = System.currentTimeMillis();
// receive our message
String filter = "JMSCorrelationID = '" + correlationId + "'";
QueueReceiver receiver = session.createReceiver(destination, filter);
TextMessage m = (TextMessage) receiver.receive();
System.out.println("Received message = " + m.getText() + " timestamp=" + m.getJMSTimestamp());
System.out.println("End listen " + getQueueName() + " " + correlationId +" Queue from receivemessage");
session.close();
connection.close();
} catch (JMSException ex) {
System.out.println("Error = " + ex.getMessage());
}
}
public static void main(String args[]) {
HelloProducerConsumer.sendMessage(getQueueName());
String correlationId1 = getCorrelationId();
HelloProducerConsumer.sendMessage(getQueueName());
String correlationId2 = getCorrelationId();
HelloProducerConsumer.sendMessage(getQueueName());
String correlationId3 = getCorrelationId();
HelloProducerConsumer.receivemessage(correlationId2);
HelloProducerConsumer.receivemessage(correlationId1);
HelloProducerConsumer.receivemessage(correlationId3);
}
}
String filter = "JMSCorrelationID = '" + msg.getJMSMessageID() + "'";
QueueReceiver receiver = session.createReceiver(queue, filter);
Qui il destinatario riceverà i messaggi per i quali JMSCorrelationID
è uguale a MessageID
. questo è molto utile nel paradigma richiesta / risposta.
oppure puoi impostarlo direttamente su qualsiasi valore:
QueueReceiver receiver = session.createReceiver(queue, "JMSCorrelationID ='"+id+"'";);
Di quanto puoi fare receiver.receive (2000);
o receiver.setMessageListener(this);