随笔 - 100  文章 - 50  trackbacks - 0
<2012年3月>
日一二三四五六
26272829123
45678910
11121314151617
18192021222324
25262728293031
1234567

常用链接

留言簿(3)

随笔分类

随笔档案

文章分类

文章档案

收藏夹

我收藏的一些文章!

搜索

  •  

最新评论

阅读排行榜

评论排行榜

与MQ通讯的完整JAVA程序
这份程序的原始文档是来自于互联网，不过不知道作者是谁，在些先谢了，我增加了个人的理解在里面，增加注释，这样便于阅读与理解。

该程序实现了发送消息与读取消息的功能，见其中的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地址及端口、本地传输队例的名称、以及本地的通信地址，这样才能够往远程主机发送消息。

[java] view plaincopyprint?
  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. }  
posted on 2012-03-04 23:15 fly 阅读(3990) 评论(0)  编辑  收藏

只有注册用户登录后才能发表评论。
网站导航:
 
Copyright ©2024 fly Powered by: 博客园 模板提供：沪江博客