Слушатель сообщений IBM MQ

Question

Привет, кто-нибудь знает, как создать прослушиватель сообщений с использованием IBM MQ? Я знаю, как это сделать, используя спецификацию JMS, но я не уверен, как это сделать для IBM MQ. Любые ссылки или указатели приветствуются.

Answer 1

Несмотря на то, что, как отмечалось предыдущими респондентами, есть Java API для WMQ, WMQ также поддерживает JMS, поэтому здесь приведены некоторые ресурсы, с которых можно начать.

Взгляните на эту статью: Технический журнал IBM WebSphere Developer: Запуск автономного приложения Java в WebSphere MQ V6.0

Кроме того, если вы установили полный клиент WMQ, а не просто взяли банки, у вас будет множество образцов кода. По умолчанию они будут находиться в C: \ Program Files \ IBM \ WebSphere MQ \ tools \ jms или / opt / mqm / samp, в зависимости от вашей платформы.

Если вам нужен установочный носитель клиента WMQ, получите его здесь . Обратите внимание, что это клиент WMQ v7, а не клиент v6. Он совместим с v6 QMgr, но по состоянию на сентябрь 2011 года v6 устарел, вам следует заняться новыми разработками для клиента v7 и, если возможно, для v7 QMgr. Если обе стороны v7, доступно множество улучшений функциональности и производительности.

Вы можете получить руководство по продукту здесь , если вам это нужно.

Наконец, убедитесь, что вы получили исключение JMS для печати связанного исключения. Это не вещь WMQ, а вещь JMS. Sun предоставила многоуровневую структуру данных для исключений JMS, и действительно интересные части часто находятся на уровне вложенности. Это не имеет большого значения и может быть реализовано в несколько строк:

try {
  .
  . code that might throw a JMSException
  .
} catch (JMSException je) {
  System.err.println("caught "+je);
  Exception e = je.getLinkedException();
  if (e != null) {
    System.err.println("linked exception: "+e);
  } else {
    System.err.println("No linked exception found.");
  }
}

Это помогает определить разницу между ошибкой JMS и ошибкой транспорта. Например, ошибкой безопасности JMS может быть WMQ 2035, или это может быть конфигурация JSSE, или приложение может не иметь доступа к чему-либо в файловой системе. Только один из них стоит потратить много времени на копание журналов ошибок WMQ, и только распечатав связанное исключение, вы сможете определить, так ли это.

Answer 2

Посмотрите справку IBM: Написание базовых Java-приложений WebSphere MQ

У IBM есть API для взаимодействия с очередями. Вот их образец:

import com.ibm.mq.*;            // Include the WebSphere MQ classes for Java package


public class MQSample
{
  private String qManager = "your_Q_manager";  // define name of queue
                                               // manager to connect to.
  private MQQueueManager qMgr;                 // define a queue manager
                                               // object
  public static void main(String args[]) {
     new MQSample();
  }

  public MQSample() {
   try {

      // Create a connection to the queue manager

      qMgr = new MQQueueManager(qManager);

      // Set up the options on the queue we wish to open...
      // Note. All WebSphere MQ Options are prefixed with MQC in Java.

      int openOptions = MQC.MQOO_INPUT_AS_Q_DEF |
                        MQC.MQOO_OUTPUT ;

      // Now specify the queue that we wish to open,
      // and the open options...

      MQQueue system_default_local_queue =
              qMgr.accessQueue("SYSTEM.DEFAULT.LOCAL.QUEUE",
                               openOptions);

      // Define a simple WebSphere MQ message, and write some text in UTF format..

      MQMessage hello_world = new MQMessage();
      hello_world.writeUTF("Hello World!");

      // specify the message options...

      MQPutMessageOptions pmo = new MQPutMessageOptions(); // accept the // defaults,
                                                           // same as MQPMO_DEFAULT

      // put the message on the queue

      system_default_local_queue.put(hello_world,pmo);

      // get the message back again...
      // First define a WebSphere MQ message buffer to receive the message into..

      MQMessage retrievedMessage = new MQMessage();
      retrievedMessage.messageId = hello_world.messageId;

      // Set the get message options...

      MQGetMessageOptions gmo = new MQGetMessageOptions(); // accept the defaults
                                                           // same as  MQGMO_DEFAULT
      // get the message off the queue...

      system_default_local_queue.get(retrievedMessage, gmo);

      // And prove we have the message by displaying the UTF message text

      String msgText = retrievedMessage.readUTF();
      System.out.println("The message is: " + msgText);
      // Close the queue...
      system_default_local_queue.close();
      // Disconnect from the queue manager

      qMgr.disconnect();
    }
      // If an error has occurred in the above, try to identify what went wrong
      // Was it a WebSphere MQ error?
    catch (MQException ex)
    {
      System.out.println("A WebSphere MQ error occurred : Completion code " +
                         ex.completionCode + " Reason code " + ex.reasonCode);
    }
      // Was it a Java buffer space error?
    catch (java.io.IOException ex)
    {
      System.out.println("An error occurred whilst writing to the message buffer: " + ex);
    }
  }
} // end of sample

Я не уверен, находятся ли банки IBM в базовом репозитории Maven. Я знаю, что в прошлом мне приходилось извлекать их из локальной установки IBM и помещать в локальное хранилище SVN. Я использую следующие банки:

<dependency>
    <groupId>com.ibm</groupId>
    <artifactId>com.ibm.mq</artifactId>
    <version>5.3.00</version>
    <scope>compile</scope>
</dependency>
    <dependency>
    <groupId>com.ibm</groupId>
    <artifactId>com.ibm.mq.pcf</artifactId>
    <version>5.3.00</version>
    <scope>compile</scope>
</dependency>
    <dependency>
    <groupId>com.ibm</groupId>
    <artifactId>com.ibm.mqbind</artifactId>
    <version>5.3.00</version>
    <scope>compile</scope>
</dependency>
    <dependency>
    <groupId>com.ibm</groupId>
    <artifactId>com.ibm.mqjms</artifactId>
    <version>5.3.00</version>
    <scope>compile</scope>
</dependency>

Answer 3

На всякий случай, если кто-нибудь запустит Google StackOverflow для MQ Listener, как я ... Это может быть не ответ из-за реализации JMS, но это то, что я искал. Примерно так:

MQQueueConnectionFactory cf = new MQQueueConnectionFactory();
MQQueueConnection conn = (MQQueueConnection)cf.createQueueConnection();
MQQueueSession session = (MQQueueSession)conn.createSession(false, 1);

Queue queue = session.createQueue("QUEUE");

MQQueueReceiver receiver = (MQQueueReceiver)session.createReceiver(queue);

receiver.setMessageListener(new YourListener());

conn.start();

YourListener должен реализовывать интерфейс MessageListener, и вы будете получать сообщения в метод onMessage (Message msg).

Answer 4

в цикле перед получением сообщения вы можете указать следующее

gmo.options = MQC.MQGMO_WAIT
gmo.waitInterval = MQConstants.MQWI_UNLIMITED;

это заставляет цикл ждать, пока в очереди не появится сообщение. Для меня это похоже на MessageListerner

Answer 5

Важный момент в дополнение к существующим ответам: JMS предоставляет MessageListener, класс, который позволяет получать сообщения в виде асинхронных обратных вызовов.

В нативном API нет эквивалента! При необходимости вы должны повторно вызывать get(...).

Answer 6

Взгляните на образец, представленный выше.

В частности, на линиях

MQGetMessageOptions gmo = new MQGetMessageOptions();       
system_default_local_queue.get(retrievedMessage, gmo);

Вы можете настроить получение для ожидания в течение указанного времени, прежде чем вызвать исключение MQRC_NO_MSG_AVAILABLE. Или ты можешь ждать вечно.

gmo.waitInterval= qTimeout;
gmo.options = MQC.MQGMO_WAIT

Таким образом, вы можете создать поток, который продолжает искать новые сообщения, а затем передавать их обработчику. Получение и установка не должны быть в одном потоке или даже в приложении.

Надеюсь, это поможет ответить на ваш вопрос.

Answer 7

Здравствуйте, вот рабочий пример прослушивателя сообщений с IBM MQ. Здесь я использовал Spring также для создания бобов и т.д ...

package queue.app;

import javax.annotation.PostConstruct;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.Queue;
import javax.jms.QueueConnection;
import javax.jms.QueueReceiver;
import javax.jms.QueueSession;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import com.ibm.mq.jms.MQQueue;
import com.ibm.mq.jms.MQQueueConnectionFactory;
import com.ibm.msg.client.wmq.WMQConstants;


@Component
public class QueueConsumer implements MessageListener{

    private Logger logger = Logger.getLogger(getClass());

    MQQueueConnectionFactory qcf = new MQQueueConnectionFactory();
    QueueConnection qc;
    Queue queue;
    QueueSession queueSession;
    QueueReceiver qr;

    @Value("${jms.hostName}")
    String jmsHost;
    @Value("${jms.port}")
    String jmsPort;
    @Value("${jms.queue.name}")
    String QUEUE_NAME;
    @Value("${jms.queueManager}")
    String jmsQueueMgr;
    @Value("${jms.username}")
    String jmsUserName;
    @Value("${jms.channel}")
    String jmsChannel;

    @PostConstruct
    public void init() throws Exception{
        qcf.setHostName (jmsHost);
        qcf.setPort (Integer.parseInt(jmsPort));
        qcf.setQueueManager (jmsQueueMgr);
        qcf.setChannel (jmsChannel);
        qcf.setTransportType (WMQConstants.WMQ_CM_CLIENT);
        qc = qcf.createQueueConnection ();

        queue = new MQQueue(QUEUE_NAME);
        qc.createQueueSession (false, Session.AUTO_ACKNOWLEDGE);
        queueSession = qc.createQueueSession (false, Session.AUTO_ACKNOWLEDGE);
        qr = queueSession.createReceiver(queue);
        qr.setMessageListener(this);
        qc.start();

    }


    @Override
    public void onMessage(Message message) {
        logger.info("Inside On Message...");
        long t1 = System.currentTimeMillis();
        logger.info("Message consumed at ...."+t1);

        try{
            if(message instanceof TextMessage) {
                logger.info("String message recieved >> "+((TextMessage) message).getText());
            }

        }catch(Exception e){
            e.printStackTrace();
        }

    }
}

Ниже приведены зависимости, которые у меня есть.

<dependency>
            <groupId>com.sun.messaging.mq</groupId>
            <artifactId>fscontext</artifactId>
            <version>4.2</version>
            <scope>test</scope>
        </dependency>

        <dependency>
            <groupId>com.ibm</groupId>
            <artifactId>jms</artifactId>
            <version>1.0</version>
        </dependency>

        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-jms</artifactId>
            <version>3.2.17.RELEASE</version>
        </dependency>


        <dependency>
            <groupId>com.ibm</groupId>
            <artifactId>com.ibm.mq</artifactId>
            <version>1.0</version>
        </dependency>
        <dependency>
            <groupId>com.ibm</groupId>
            <artifactId>com.ibm.mq.allclient</artifactId>
            <version>1.0</version>
        </dependency>
        <dependency>
            <groupId>com.ibm</groupId>
            <artifactId>com.ibm.mq.jmqi</artifactId>
            <version>1.0</version>
        </dependency>
        <dependency>
            <groupId>com.ibm</groupId>
            <artifactId>com.ibm.mqjms</artifactId>
            <version>1.0</version>
        </dependency>