asyncmessageconsumer.java

J2EE开发与Weblogic一书中的源代码

package com.learnweblogic.examples.ch7.queue;

import com.learnweblogic.examples.BaseClient;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.Queue;
import javax.jms.QueueConnection;
import javax.jms.QueueConnectionFactory;
import javax.jms.QueueReceiver;
import javax.jms.QueueSession;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.naming.Context;
import javax.naming.NamingException;

public final class AsyncMessageConsumer 
  extends BaseClient 
  implements MessageListener
{

  private int EXPECTED_MESSAGE_COUNT = 2;
  private int messageCount = 0;

  private QueueReceiver receiver;
  private TextMessage msg;

  public AsyncMessageConsumer(String [] argv) 
    throws JMSException, NamingException
  {
    super(argv);

    Context ctx = getInitialContext();

    QueueConnectionFactory qConFactory = (QueueConnectionFactory)
      ctx.lookup("weblogic.jms.ConnectionFactory"); 

    Queue messageQueue = (Queue) ctx.lookup("MessageQueue");

    QueueConnection qCon = qConFactory.createQueueConnection();

    QueueSession session = qCon.createQueueSession(
      false, /* not a transacted session */
      Session.AUTO_ACKNOWLEDGE 
    );

    receiver = session.createReceiver(messageQueue);

    receiver.setMessageListener(this);

    qCon.start();

  }

  public boolean expectMoreMessages() { 
    return messageCount < EXPECTED_MESSAGE_COUNT;
  }

  public void onMessage(Message m) {

    try {
      TextMessage msg = (TextMessage) m;

      System.err.println("Received: "+msg.getText());

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

    messageCount++;
  }



  public static void main(String [] argv) 
    throws Exception
  {
    int MAX_TRIES = 10;
    int tryCount = 0;

    AsyncMessageConsumer consumer = new AsyncMessageConsumer(argv);

    while (consumer.expectMoreMessages() && (tryCount < MAX_TRIES)) {
      try {
        Thread.sleep(1000);
      } catch (InterruptedException ignore) {}

      tryCount++;
    }
  }

}