Filtragem do receptor de mensagem JMS por JMSCorreLationId
Pergunta
Como posso instanciar um ouvinte da fila JMS em Java (JRE / JDK / J2EE 1.4) que recebe apenas mensagens que correspondem a um determinado JMSCorreLationId? As mensagens que estou procurando pegar foram publicadas em uma fila e não um tópico, embora isso possa mudar, se necessário.
Aqui está o código que estou usando atualmente para colocar a mensagem na fila:/**
* 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
Solução
A configuração da conexão da fila é a mesma, mas depois de ter a fila, você define o seletor ao criar um receptor.
QueueReceiver receiver = session.createReceiver(myQueue, "JMSCorrelationID='theid'");
então
receiver.receive()
ou
receiver.setListener(myListener);
Outras dicas
Btw enquanto não é a pergunta real que você fez - se você está tentando implementar a resposta da solicitação sobre o JMS, eu recomendo lendo este artigo Como a API JMS é um pouco mais complexa do que você pode imaginar e fazer isso com eficiência é muito mais difícil do que parece.
Em particular Para usar o JMS com eficiência Você deve tentar evitar a criação de consumidores para uma única mensagem etc.
Além disso, porque a API JMS é muito complexa de usar de maneira correta e eficiente - principalmente com agrupamento, transações e processamento simultâneo - eu recomendo pessoas esconder o middleware do código do aplicativo como via usar Implementação de Spring Remoting da Apache Camel para JMS
Espero que isso ajude. Eu usei MQ aberto.
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);
Aqui o receptor receberá as mensagens para as quais JMSCorrelationID
é igual a MessageID
. Isso é muito útil no paradigma de solicitação/ resposta.
Ou você pode definir diretamente isso para qualquer valor:
QueueReceiver receiver = session.createReceiver(queue, "JMSCorrelationID ='"+id+"'";);
Do que você pode fazer receiver.receive(2000);
ou receiver.setMessageListener(this);