protocol ejemplo curso colas cola java-ee queue jms

java ee - ejemplo - Filtrado del receptor de mensajes JMS por JMSCorrelationID



jms protocol (4)

¿Cómo puedo crear una instancia de un detector de colas JMS en java (JRE / JDK / J2EE 1.4) que solo recibe mensajes que coinciden con un determinado JMSCorrelationID? Los mensajes que estoy buscando recoger se han publicado en una cola y no en un tema, aunque eso puede cambiar si es necesario.

Aquí está el código que estoy usando para poner el mensaje en la cola:

/** * 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


La configuración de la conexión de cola es la misma, pero una vez que tiene la QueueSession, configura el selector al crear un receptor.

QueueReceiver receiver = session.createReceiver(myQueue, "JMSCorrelationID=''theid''");

entonces

receiver.receive()

o

receiver.setListener(myListener);


Por cierto, aunque no es la pregunta real que hiciste: si intentas implementar la respuesta de solicitud sobre JMS, te recomiendo leer este artículo ya que la API de JMS es bastante más compleja de lo que puedas imaginar y hacer esto de manera eficiente es mucho más difícil que mira

En particular, para usar JMS de manera eficiente , debe intentar evitar la creación de consumidores para un solo mensaje, etc.

Además, debido a que la API de JMS es muy compleja de usar correcta y eficientemente, particularmente con la agrupación, las transacciones y el procesamiento simultáneo, recomiendo a las personas que oculten el middleware de su código de aplicación , como mediante la implementación de Apache Camel''s Spring Remoting para JMS


String filter = "JMSCorrelationID = ''" + msg.getJMSMessageID() + "''"; QueueReceiver receiver = session.createReceiver(queue, filter);

Aquí el receptor obtendrá los mensajes para los cuales JMSCorrelationID es igual a MessageID . esto es muy útil en el paradigma de solicitud / respuesta.

o puede establecerlo directamente a cualquier valor:

QueueReceiver receiver = session.createReceiver(queue, "JMSCorrelationID =''"+id+"''";);

Que puedes hacer con cualquiera de los receiver.receive(2000); o receiver.setMessageListener(this);


Espero que esto ayude Usé 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); } }