تصفية متلقي رسائل JMS بواسطة JMSCorrelationID

Question

كيف يمكنني إنشاء مثيل لمستمع قائمة انتظار JMS في Java (JRE /JDK / J2EE 1.4) الذي يتلقى فقط الرسائل التي تطابق معرف JMSCorrelationID المحدد؟تم نشر الرسائل التي أتطلع إلى التقاطها في قائمة انتظار وليس في موضوع، على الرغم من أن ذلك يمكن أن يتغير إذا لزم الأمر.

إليك الرمز الذي أستخدمه حاليًا لوضع الرسالة في قائمة الانتظار:

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

Answer 1

إعداد اتصال قائمة الانتظار هو نفسه، ولكن بمجرد حصولك على QueueSession، يمكنك تعيين المحدد عند إنشاء جهاز استقبال.

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

ثم

receiver.receive()

أو

receiver.setListener(myListener);

Answer 2

راجع للشغل في حين أنه ليس السؤال الفعلي الذي طرحته - إذا كنت تحاول تنفيذ استجابة الطلب عبر JMS فأنا أوصي به قراءة هذا المقال نظرًا لأن واجهة برمجة تطبيقات JMS أكثر تعقيدًا مما قد تتخيله، كما أن القيام بذلك بكفاءة أصعب بكثير مما يبدو.

بخاصة لاستخدام JMS بكفاءة يجب أن تحاول تجنب إنشاء مستهلكين لرسالة واحدة وما إلى ذلك.

وأيضًا نظرًا لأن واجهة برمجة تطبيقات JMS معقدة للغاية لاستخدامها بشكل صحيح وفعال - خاصة مع التجميع والمعاملات والمعالجة المتزامنة - أوصي بالأشخاص إخفاء البرامج الوسيطة من رمز التطبيق الخاص بهم مثل عن طريق استخدام تطبيق Apache Camel's Spring Remoting لـ JMS

Answer 3

نأمل أن يساعد هذا.لقد استخدمت 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);
}
}

Answer 4

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

هنا سوف يحصل المتلقي على الرسائل التي JMSCorrelationID مساوي ل MessageID.وهذا مفيد جدًا في نموذج الطلب/الاستجابة.

أو يمكنك ضبط هذا مباشرة على أي قيمة:

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

مما يمكنك القيام به سواء receiver.receive(2000); أو receiver.setMessageListener(this);