traderreceive.java

weblogic应用全实例

//声明这个类定义在包examples.jms.trader中
package examples.jms.trader;
//声明这个类引入的其它包或类
import java.io.*;
import java.rmi.RemoteException;
import java.util.*;
import javax.ejb.*;
import javax.jms.*;
import javax.transaction.*;
import javax.naming.*;
import examples.ejb.basic.statelessSession.*;

/**
 * 接收消息并调用EJBean来处理
 */
public class TraderReceive
{
/**
 * 定义JNDI上下文
 */
  public final static String JNDI_FACTORY="weblogic.jndi.WLInitialContextFactory";
/**
 *  定义JMS连接构造器
 */
  public final static String JMS_FACTORY="jms.connection.traderFactory";
/**
 * 定义EJB无状态会话名
 */
  public final static String EJB_HOME="statelessSession.TraderHome";

/**
 * 定义JMS主题名
 */
  public final static String TOPIC="weblogic.examples.jms.exampleTopic";
/**
 * 定义JTA用户事务
 */
  public final static String TX="javax.transaction.UserTransaction";
  //声明主题连接构造器
  private TopicConnectionFactory connectionFactory;
  //声明主题连接
  private TopicConnection connection;
  //声明主题连接会话
  private TopicSession session;
  //声明主题发布器
  private TopicSubscriber subscriber;
  //声明主题
  private Topic topic;
  //声明远程EJB
  private Trader ejbTrader;
  //声明用户事务
  private UserTransaction tx;

  /** 
    * 消息监听器接口
   */
  public void processMessages()
  {
    while(true) {
      try {
        System.out.println("Waiting to receive message...");
        //等待接收消息
        Message msg = subscriber.receive();

        if (msg instanceof MapMessage) {
        //是映射消息
          Exception e = null;
          try {
          //事务定义开始
            tx.begin();
            MapMessage m = (MapMessage) msg;
            //获取顾客名
            String customerName = m.getString("CustomerName"); 
            //获取顾客类型
            String tradeType = m.getString("TradeType");
            //获取股票类型
            String symbol = m.getString("Symbol"); 
            //获取交易数
            int numberOfShares = m.getInt("Shares"); 
            try {
              TradeResult tr = null;
            
              if ("buy".equalsIgnoreCase(tradeType)) {
              //如果是买
              //买
                tr = ejbTrader.buy(symbol, numberOfShares);
                System.out.println("Bought " + tr.getNumberTraded());
               //事务执行
                tx.commit();
              } else {
              
                if ("sell".equalsIgnoreCase(tradeType)) {
                //如果是卖
                //卖
                  tr = ejbTrader.sell(symbol, numberOfShares);
                  System.out.println("Sold " + tr.getNumberTraded());
                  //事务执行
                  tx.commit();
                } else {
                  System.out.println("Rolling Back Transaction");
                  tx.rollback();
                  System.out.println("Unknown TradeType: "+tradeType);
                }
              }
            } catch (RollbackException re) {
              e = re;
            } catch (HeuristicRollbackException hre) {
              e = hre;
            } catch (HeuristicMixedException hme) {
              e = hme;
            } catch (RemoteException re) {
              e = re;
            }
            if (e != null) {
              e.printStackTrace();
            }
          } catch (SystemException se) {
            e = se;
          } catch (NotSupportedException nse) {
            nse.printStackTrace();
          }
        }
      } catch (JMSException jmse) {
        try {
          tx.rollback();
        } catch (SystemException se) {}
        jmse.printStackTrace();
      }
    }
  }

  /**
   * 初始化方法，创建所有必要的对象
   */
  public void init(Context ctx, String topicName)
       throws NamingException, JMSException, RemoteException, CreateException
  {
  //查找连接构造器
    connectionFactory = (TopicConnectionFactory) ctx.lookup(JMS_FACTORY);
   //创建连接对象
    connection = connectionFactory.createTopicConnection();
    //设置客户端Id
    connection.setClientID("traderReceive");
    //创建会话
    session = connection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
    //查找主题
    topic = (Topic) ctx.lookup(topicName);
    //创建订阅
    subscriber = session.createDurableSubscriber(topic, "traderReceive");
    //查找EJB主接口
    TraderHome brokerage = (TraderHome) ctx.lookup(EJB_HOME);
    //创建EJB
    ejbTrader = brokerage.create();
    //查找事务对象
    tx = (javax.transaction.UserTransaction) ctx.lookup(TX);
    //启动连接
    connection.start();
  }

  /**
   * 关闭JMS对象
   */
  public void close()
       throws JMSException
  {
  //关闭订阅，会话，连接
    subscriber.close();
    session.close();
    connection.close();
  }

  /**
   * main()方法
   */

  public static void main(String[] args)
       throws Exception 
  {
    if (args.length != 1) {
      System.out.println("用法: java examples.jms.trader.TraderReceive WebLogicURL");
      return;
    }
    //获取名称服务上下文
    InitialContext ic = getInitialContext(args[0]);
    //本类实例
    TraderReceive demo = new TraderReceive();
    //调用本类的初始化方法
    demo.init(ic, TOPIC);
    //调用本类的示例方法
    demo.processMessages();
    //关闭打开的对象
    demo.close();
  }

  private static InitialContext getInitialContext(String url)
       throws NamingException
  {
    Hashtable env = new Hashtable();
    env.put(Context.INITIAL_CONTEXT_FACTORY, JNDI_FACTORY);
    env.put(Context.PROVIDER_URL, url);
    return new InitialContext(env);
  }

}