[原创]JMS队列收发消息实例_Tomcat, WebLogic及J2EE讨论区_Weblogic技术|Tuxedo技术|中间件技术|Oracle论坛|JAVA论坛|Linux/Unix技术|hadoop论坛

总帖数

每页帖数

1/1页

返回列表

发起投票

查看: 3182 | 回复: 0

主题： [原创]JMS队列收发消息实例

rui.yuan

注册用户

等级：中校
经验：1637
发帖：29
精华：0
注册：1970-1-1
状态：离线
发送短消息息给rui.yuan

加好友发送短消息息给rui.yuan

发消息

发表于：

2014-11-27 13:15:45 | [全部帖] [楼主帖]

楼主

首先确定已经配置好JMS

打开weblogic的控制台，进入到JMS的监控页

北京联动北方科技有限公司

因为不管接收消息还会发送消息都需获取context，所以把它写成了一个帮助类，代码如下：

package com.landing.util;
import java.util.Properties;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;
public class ContextManager {
      public static Context getContext(){
            Context ctx = null;
            Properties pro = new Properties();
            pro.put(Context.INITIAL_CONTEXT_FACTORY, "weblogic.jndi.WLInitialContextFactory");
            pro.put(Context.PROVIDER_URL, "t3://localhost:7001");
            try {
                  ctx = new InitialContext(pro);
            } catch (NamingException e) {
                  e.printStackTrace();
            }
            return ctx;
      }
}

发送消息代码：

package com.landing.queue;
import javax.jms.*;
import javax.naming.Context;
import javax.naming.NamingException;
import com.landing.util.ContextManager;
public class SendMessage {
      private QueueSender sender;
      private TextMessage msg;
      public static void main(String[] args) {
            SendMessage sm = new SendMessage(args);
            sm.runClient("yuanrui");
            System.out.println("successful!");
      }
      public SendMessage(String[] argv){
            Context ctx = null;
            try {
                  ctx =ContextManager.getContext();
                  QueueConnectionFactory qcf = (QueueConnectionFactory) ctx.lookup("mConnectionFactory");
                  QueueConnection qc = qcf.createQueueConnection();
                  QueueSession qsession = qc.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
                  Queue que = (Queue) ctx.lookup("mqueue");
                  sender = qsession.createSender(que);
                  msg = qsession.createTextMessage();
            } catch (NamingException e) {
                  e.printStackTrace();
            } catch (JMSException e) {
                  e.printStackTrace();
            }
      }
      public void runClient(String str){
            try {
                  msg.setText("hello world");
                  sender.send(msg);
                  msg.setText("success conn to JMS");
                  sender.send(msg);
                  msg.setText(str);
                  sender.send(msg);
            } catch (JMSException e) {
                  e.printStackTrace();
            }
      }
}

为了完成同步和异步的接收实验，发送运行了三次，共计9条消息记录

发送完后，可以看到监控上当前消息数变成9，如下图：

北京联动北方科技有限公司

同步接收消息代码：

package com.landing.queue;
import javax.jms.*;
import javax.naming.Context;
import javax.naming.NamingException;
import com.landing.util.ContextManager;
public class ReceiveMessage {
      private QueueReceiver receiver;
      private TextMessage msg;
      public static void main(String[] args) throws JMSException {
            ReceiveMessage rm = new ReceiveMessage();
            rm.runClient();
      }
      public ReceiveMessage (){
            Context ctx = null;
            try {
                  ctx = ContextManager.getContext();
                  QueueConnectionFactory qcf = (QueueConnectionFactory) ctx.lookup("mConnectionFactory");
                  QueueConnection qc = qcf.createQueueConnection();
                  qc.start();
                  QueueSession qsession = qc.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
                  Queue que = (Queue) ctx.lookup("mqueue");
                  receiver = qsession.createReceiver(que);
                  qc.start();
            } catch (NamingException e) {
                  e.printStackTrace();
            } catch (JMSException e) {
                  e.printStackTrace();
            }
      }
      public void runClient() throws JMSException{
            msg = (TextMessage) receiver.receive();
            System.out.println("receive:"+msg.getText());
            msg = (TextMessage) receiver.receive();
            System.out.println("receive:"+msg.getText());
            msg = (TextMessage) receiver.receive();
            System.out.println("receive:"+msg.getText());
      }
}

myEclipse上运行结果如下：

北京联动北方科技有限公司

weblogic上监控如下：

北京联动北方科技有限公司

异步接收消息代码：

package com.landing.queuet;
import javax.jms.*;
import javax.naming.Context;
import javax.naming.NamingException;
import com.landing.util.ContextManager;
public class AsynReceiveMessage implements MessageListener{
      private int execeptMessCount = 2;
      private int messageCount = 0;
      private QueueReceiver receiver;
      private TextMessage msg;
      public static void main(String[] args) {
            int maxNum = 10;
            int tryCount = 0;
            AsynReceiveMessage arm = new AsynReceiveMessage();
            while(arm.exceptMoreMessage()&&tryCount<maxNum){
                  try {
                        Thread.sleep(1000);
                  } catch (InterruptedException e) {
                        e.printStackTrace();
                  }
                  tryCount++;
            }
            System.out.println("******************");
      }
      public AsynReceiveMessage(){
            Context ctx = null;
            try {
                  ctx = ContextManager.getContext();
                  QueueConnectionFactory qcf = (QueueConnectionFactory) ctx.lookup("mConnectionFactory");
                  QueueConnection qconn = qcf.createQueueConnection();
                  QueueSession qsession = qconn.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
                  Queue queue = (Queue) ctx.lookup("mqueue");
                  receiver = qsession.createReceiver(queue);
                  receiver.setMessageListener(this);
                  qconn.start();
            } catch (NamingException e) {
                  e.printStackTrace();
            } catch (JMSException e) {
                  e.printStackTrace();
            }
      }
      public void onMessage(Message message) {
            msg = (TextMessage) message;
            System.out.println("reveive asynMess:"+msg);
            messageCount++;
      }
      public boolean exceptMoreMessage(){
            return messageCount<execeptMessCount;
      }
}

myEclipse上运行结果如下：

北京联动北方科技有限公司