消息接收的多线程实现

package duoxianchengtest;

import java.util.Properties;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.ObjectMessage;
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.InitialContext;
import javax.naming.NamingException;
/**
 *
 * @author Owner
 *
 *
 */

public class Receiver1
{

   QueueReceiver qrec1=null;
   QueueSession qs =null;
   QueueConnection qc=null;
   /**
    *
    * @throws NamingException
    * @throws JMSException
    * 构造函数初始化
    */
public Receiver1() throws NamingException, JMSException{
//  ictx = new InitialContext(JMSConstant.getProp());
  Properties pr =new Properties();
  pr.put("82.0.176.214","16400");
  Context ictx = new InitialContext(pr);
     Queue queue = (Queue) ictx.lookup("queue");
     QueueConnectionFactory qcf = (QueueConnectionFactory) ictx.lookup("qcf");
     ictx.close();
     qc = qcf.createQueueConnection("user1","user1");
     qs = qc.createQueueSession(false,Session.AUTO_ACKNOWLEDGE);
     qrec1 = qs.createReceiver(queue,"sessionID=2");
}
/**
 *
 * @author Owner
 *  一个内部类继承Thread实现run()方法,在run()方法设置消息监听器.
 */
  class Brun extends Thread{
  public void run(){
   try{
    qrec1.setMessageListener(new MsgListener1());
    qc.start();
       }catch(JMSException e){}
         }
       }
  /**
   * 创建线程组
   *
   */
  public void run_thread() {
  Brun[] tt = new Brun[3000];
  for (int i = 0; i < tt.length; i++) {
   tt[i] = new Brun();
   tt[i].start();

   // System.out.println("go");
  }

  // for (int i = 0; i < tt.length; i++) {
  // System.out.println("go1");
  // try {
  // tt[i].join();
  //
  // } catch (InterruptedException je) {}
  //   
  //   
  // System.out.println("go2");
  //  }
 }
     
 
 void close() throws JMSException {
   qrec1.close();
  qs.close();
  qc.close();
 }
 /**
  *
  * @param args
  * @throws Exception
  */
  public static void main(String[] args) throws Exception
  {
//  long time1= System.currentTimeMillis();
   
    System.out.println("Requests to receive messages...");

      Receiver1 rc=new Receiver1();
      rc.run_thread();
      System.in .read();
      rc.close();
      System.out.println();
      System.out.println( " messages received.");
//    long time2 = System.currentTimeMillis();
//    System.out.println("time=:"+(time2-time1));
   
  }
}

本来也想做个join()的.但是只是设置消息监听器完毕.不行呀..看来要关闭连接对象,需要被动的接收消息,,然后根据消息内容来处理了关闭了?还有一个问题是.线程是否稳定.在ejb里消息驱动bean是不支持多线程发送的.一切都还需要测试.

评论

3 楼 dev_liu 2007-01-05  

我发现自己太喜欢多线程了..什么时候都在用,这个接收的客户端就嵌套了2个多线程.一个设置消息监听器,一个是对消息的getText()的动作.

2 楼 dev_liu 2007-01-05  

package duoxianchengtest;

import java.util.Date;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.TextMessage;





public class Test extends Thread{
TextMessage msg ;

public Test(TextMessage  msg){
this.msg= msg;

}
public void run() {
Date date = new Date();

    System.out.println("time1:" + date.getTime());
try { 
System.out.println("test"+msg.getText() );
} catch (JMSException e) {
// TODO 自动生成 catch 块
e.printStackTrace();
}
    


}
}

1 楼 dev_liu 2007-01-05  

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.ObjectMessage;
import javax.jms.TextMessage;





public class MsgListener1 implements MessageListener
{

      TextMessage kmsg =null;

  public void onMessage(Message msg) {

  if (msg instanceof TextMessage)
    kmsg=(TextMessage)msg;
    new Test(kmsg).start();
}

}