asyncmessageconsumer.java

java的一系列产品中包括jsme,jmse,j2ee,本文件提供j2ee实现的源代码.

/*异步消息接收程序，从消息队列接收*/
package queue;

import java.util.Properties;
import javax.rmi.PortableRemoteObject;
import javax.naming.InitialContext;
import javax.naming.Context;
import javax.naming.NamingException;

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;

public final class AsyncMessageConsumer implements MessageListener
{
  private static QueueReceiver receiver;
  private static TextMessage msg;  
  private static final String url  = "t3://localhost:7001";  
  
  static int EXPECTED_MESSAGE_COUNT = 2;
  static int messageCount = 0;
  
  public static void main(String [] argv) throws Exception
  {   
    int MAX_TRIES = 10;
    int tryCount = 0;
    
    try{
     	//设置初始上下文
      Properties env = new Properties();
      env.put(Context.INITIAL_CONTEXT_FACTORY,"weblogic.jndi.WLInitialContextFactory");
      env.put(Context.PROVIDER_URL, url);
      
    	//获取初始上下文      
    	Context ctx = new InitialContext(env);    	
    	
    	//1：通过JDNI查找ConnectionFactory
    	QueueConnectionFactory qConFactory = (QueueConnectionFactory)
      		ctx.lookup("weblogic.jms.ConnectionFactory"); 
      
    	//2：用ConnectionFactory建立JMS连接
    	QueueConnection qCon = qConFactory.createQueueConnection();
      
      //3：用Connection创建会话
    	QueueSession session = qCon.createQueueSession(
      	false, /* not a transacted session */
      	Session.AUTO_ACKNOWLEDGE 
    	);
    	
    	//4：通过JDNI查找目的（消息队列）
    	Queue messageQueue = (Queue) ctx.lookup("MessageQueue");
      
      //5：创建消息接收者，并使用setMessageListener()方法传递一个实现了
      //MessageListener接口的对象
    	receiver = session.createReceiver(messageQueue);
    	receiver.setMessageListener(new AsyncMessageConsumer());
    	
    	//异步消息接收者：start()方法调用onMessage()接收消息
    	qCon.start();
    	
      //为避免主线程在消息到达之前退出，执行以下循环
      //while(true) {
      while(expectMoreMessages()&&(tryCount<MAX_TRIES)) {
      	try {
        	Thread.sleep(1000);
      	} catch (InterruptedException ignore) {}
      	tryCount++;   	
      }
    }catch (JMSException e) {
      e.printStackTrace();
    }
  }
  
  public static boolean expectMoreMessages(){ 
    return messageCount < EXPECTED_MESSAGE_COUNT;
  }
  
  public void onMessage(Message m) {
    //6：创建文本消息，并接收
    try {
      TextMessage msg = (TextMessage) m;
      System.out.println("Received: "+msg.getText());
      System.out.println("Received: "+msg.getStringProperty("MyProp"));
    } catch (JMSException e) {
      e.printStackTrace();
    }     
  }
}