注册 登录  
 加关注
查看详情
   显示下一条  |  关闭
温馨提示!由于新浪微博认证机制调整,您的新浪微博帐号绑定已过期,请重新绑定!立即重新绑定新浪微博》  |  关闭

隐龙 为了一生的信念

今日默默沉于水,他日飞腾在九天...

 
 
 

日志

 
 

与MQ通讯的完整JAVA程序  

2013-02-22 11:33:01|  分类: JAVA |  标签: |举报 |字号 订阅

  下载LOFTER 我的照片书  |
这份程序的原始文档是来自于互联网,不过不知道作者是谁,在些先谢了,我增加了个人的理解在里面,增加注释,这样便于阅读与理解。

该程序实现了发送消息与读取消息的功能,见其中的send***与get***方法。这只适合于测试,因为环境中的程序还需要对此有稍微的更改,在真实的环境中肯定是在while(true){...}的无限循环中去调用其中的get方法,如果有值,那就执行对消息的处理操作,如果没有值就继续循环,在get方法中有等待的时间。

这个程序就其本身来说还是比较理解的:

1、首先设置一些相关的环境变量

2、再连接队列管理器

3、再次操作队列管理器中的指定队列

4、往指定队列中发消息或者是从指定对列中取消息

5、关闭队列

如果不知道如何在MQ资源管理器中配置远程队列及通过远程队列往远程的MQ发送消息,请参见文章:

http://blog.csdn.net/fenglibing/archive/2009/05/08/4160639.aspx

真实环境中的MQ,个人觉得至少都应该有两个本地队列加一个远程队列,因为消息的交互肯定是相互的,有收消息,肯定也有发消息。一个本地队列用于接收外部发过来的消息,用法为正常;另一个本地队例用于传输,用于做于远程队例的传输队列,将消息发送给远程主机的本地队列。要使消息能够成功的传送到远程队列,还需要配置通道,通常中需要指定远程通道的IP地址及端口、本地传输队例的名称、以及本地的通信地址,这样才能够往远程主机发送消息。

  1. /** 
  2.  * @author Fenglb E-mail:56553655@163.com 
  3.  * @version 创建时间:2009-4-30 下午04:13:38 
  4.  * 类说明 
  5.  */  
  6. import java.io.IOException;  
  7. import com.ibm.mq.MQC;  
  8. import com.ibm.mq.MQEnvironment;  
  9. import com.ibm.mq.MQException;  
  10. import com.ibm.mq.MQGetMessageOptions;  
  11. import com.ibm.mq.MQMessage;  
  12. import com.ibm.mq.MQPutMessageOptions;  
  13. import com.ibm.mq.MQQueue;  
  14. import com.ibm.mq.MQQueueManager;  
  15.   
  16. public class MessageByMQ{  
  17.      //定义队列管理器和队列的名称   
  18.      private static String qmName;   
  19.      private static String qName;  
  20.      private static MQQueueManager qMgr;  
  21.      static{  
  22.          //设置环境:   
  23.          //MQEnvironment中包含控制MQQueueManager对象中的环境的构成的静态变量,MQEnvironment的值的设定会在MQQueueManager的构造函数加载的时候起作用,   
  24.          //因此必须在建立MQQueueManager对象之前设定MQEnvironment中的值.   
  25.          MQEnvironment.hostname="10.24.1.180";          //MQ服务器的IP地址         
  26.          MQEnvironment.channel="S_FENGLB";              //服务器连接的通道   
  27.          MQEnvironment.CCSID=1381;                      //服务器MQ服务使用的编码1381代表GBK、1208代表UTF(Coded Character Set Identifier:CCSID)   
  28.          MQEnvironment.port=1414;                       //MQ端口   
  29.          qmName = "QM_FENGLB";                          //MQ的队列管理器名称   
  30.          qName = "testQ";                               //MQ远程队列的名称   
  31.          try {  
  32.              //定义并初始化队列管理器对象并连接    
  33.              //MQQueueManager可以被多线程共享,但是从MQ获取信息的时候是同步的,任何时候只有一个线程可以和MQ通信。   
  34.             qMgr = new MQQueueManager(qmName);  
  35.         } catch (MQException e) {  
  36.             // TODO Auto-generated catch block   
  37.             System.out.println("初使化MQ出错");  
  38.             e.printStackTrace();  
  39.         }   
  40.      }  
  41.      /** 
  42.       * 往MQ发送消息 
  43.       * @param message 
  44.       * @return 
  45.       */  
  46.      public static int sendMessage(String message){  
  47.          int result=0;  
  48.          try{     
  49.              //设置将要连接的队列属性   
  50.              // Note. The MQC interface defines all the constants used by the WebSphere MQ Java programming interface    
  51.              //(except for completion code constants and error code constants).   
  52.              //MQOO_INPUT_AS_Q_DEF:Open the queue to get messages using the queue-defined default.   
  53.              //MQOO_OUTPUT:Open the queue to put messages.   
  54.              /*目标为远程队列,所有这里不可以用MQOO_INPUT_AS_Q_DEF属性*/  
  55.              //int openOptions = MQC.MQOO_INPUT_AS_Q_DEF | MQC.MQOO_OUTPUT;   
  56.              /*以下选项可适合远程队列与本地队列*/  
  57.              int openOptions = MQC.MQOO_OUTPUT | MQC.MQOO_FAIL_IF_QUIESCING;  
  58.              //连接队列    
  59.              //MQQueue provides inquire, set, put and get operations for WebSphere MQ queues.    
  60.              //The inquire and set capabilities are inherited from MQManagedObject.    
  61.              /*关闭了就重新打开*/  
  62.             if(qMgr==null || !qMgr.isConnected()){  
  63.                 qMgr = new MQQueueManager(qmName);  
  64.             }  
  65.              MQQueue queue = qMgr.accessQueue(qName, openOptions);            
  66.              //定义一个简单的消息   
  67.              MQMessage putMessage = new MQMessage();   
  68.              //将数据放入消息缓冲区   
  69.              putMessage.writeUTF(message);    
  70.              //设置写入消息的属性(默认属性)   
  71.              MQPutMessageOptions pmo = new MQPutMessageOptions();             
  72.              //将消息写入队列    
  73.              queue.put(putMessage,pmo);   
  74.              queue.close();  
  75.          }catch (MQException ex) {   
  76.              System.out.println("A WebSphere MQ error occurred : Completion code "   
  77.              + ex.completionCode + " Reason code " + ex.reasonCode);   
  78.              ex.printStackTrace();  
  79.          }catch (IOException ex) {   
  80.              System.out.println("An error occurred whilst writing to the message buffer: " + ex);   
  81.          }catch(Exception ex){  
  82.              ex.printStackTrace();  
  83.          }finally{  
  84.              try {  
  85.                 qMgr.disconnect();  
  86.             } catch (MQException e) {  
  87.                 e.printStackTrace();  
  88.             }  
  89.           }  
  90.          return result;  
  91.      }  
  92.      /** 
  93.       * 从队列中去获取消息,如果队列中没有消息,就会发生异常,不过没有关系,有TRY...CATCH,如果是第三方程序调用方法,如果无返回则说明无消息 
  94.       * 第三方可以将该方法放于一个无限循环的while(true){...}之中,不需要设置等待,因为在该方法内部在没有消息的时候会自动等待。 
  95.       * @return 
  96.       */  
  97.      public static String getMessage(){  
  98.          String message=null;  
  99.          try{              
  100.              //设置将要连接的队列属性   
  101.              // Note. The MQC interface defines all the constants used by the WebSphere MQ Java programming interface    
  102.              //(except for completion code constants and error code constants).   
  103.              //MQOO_INPUT_AS_Q_DEF:Open the queue to get messages using the queue-defined default.   
  104.              //MQOO_OUTPUT:Open the queue to put messages.   
  105.              int openOptions = MQC.MQOO_INPUT_AS_Q_DEF | MQC.MQOO_OUTPUT;  
  106.              MQMessage retrieve = new MQMessage();  
  107.              //设置取出消息的属性(默认属性)   
  108.              //Set the put message options.(设置放置消息选项)    
  109.              MQGetMessageOptions gmo = new MQGetMessageOptions();   
  110.              gmo.options = gmo.options + MQC.MQGMO_SYNCPOINT;//Get messages under sync point control(在同步点控制下获取消息)    
  111.              gmo.options = gmo.options + MQC.MQGMO_WAIT;  // Wait if no messages on the Queue(如果在队列上没有消息则等待)    
  112.              gmo.options = gmo.options + MQC.MQGMO_FAIL_IF_QUIESCING;// Fail if Qeue Manager Quiescing(如果队列管理器停顿则失败)    
  113.              gmo.waitInterval = 1000 ;  // Sets the time limit for the wait.(设置等待的毫秒时间限制)    
  114.              /*关闭了就重新打开*/  
  115.             if(qMgr==null || !qMgr.isConnected()){  
  116.                 qMgr = new MQQueueManager(qmName);  
  117.             }  
  118.              MQQueue queue = qMgr.accessQueue(qName, openOptions);   
  119.              // 从队列中取出消息   
  120.              queue.get(retrieve, gmo);  
  121.              message = retrieve.readUTF();    
  122.              System.out.println("The message is: " + message);   
  123.              queue.close();  
  124.          }catch (MQException ex) {   
  125.              System.out.println("A WebSphere MQ error occurred : Completion code "   
  126.              + ex.completionCode + " Reason code " + ex.reasonCode);   
  127.          }catch (IOException ex) {   
  128.              System.out.println("An error occurred whilst writing to the message buffer: " + ex);   
  129.          }catch(Exception ex){  
  130.              ex.printStackTrace();  
  131.          }finally{  
  132.              try {  
  133.                 qMgr.disconnect();  
  134.             } catch (MQException e) {  
  135.                 e.printStackTrace();  
  136.             }  
  137.          }  
  138.          return message;  
  139.      }  
  140.      public static void main(String args[]) {  
  141.          /*下面两个方法可同时使用,也可以单独使用*/  
  142.          sendMessage("this is a test");  
  143.          //getMessage();   
  144.      }  
  145. }  
  评论这张
 
阅读(525)| 评论(0)
推荐 转载

历史上的今天

评论

<#--最新日志,群博日志--> <#--推荐日志--> <#--引用记录--> <#--博主推荐--> <#--随机阅读--> <#--首页推荐--> <#--历史上的今天--> <#--被推荐日志--> <#--上一篇,下一篇--> <#-- 热度 --> <#-- 网易新闻广告 --> <#--右边模块结构--> <#--评论模块结构--> <#--引用模块结构--> <#--博主发起的投票-->
 
 
 
 
 
 
 
 
 
 
 
 
 
 

页脚

网易公司版权所有 ©1997-2018