与MQ通讯的完整JAVA程序

标签: mq 通讯 完整 | 发表时间:2016-06-11 23:42 | 作者:bijian1013
出处:http://www.iteye.com

        本文实例是基于 WebSphere MQ中将消息发送至远程队列的配置的基础上的,且如果要能正常运行并发送、接收消息,还需要在两个队列管理器(QM_ORANGE和QM_APPLE)上做如下配置或修改。

        1.创建名称为DC.SVRCONN的服务器连接通道


        2.将队列管理器的通道认证记录设置为“已禁用”


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

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

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

2.再连接队列管理器

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

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

5.关闭队列

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

        实例代码如下:

SendMessage.java

package com.bijian.study;

import java.io.IOException;
import com.ibm.mq.MQC;
import com.ibm.mq.MQEnvironment;
import com.ibm.mq.MQException;
import com.ibm.mq.MQMessage;
import com.ibm.mq.MQPutMessageOptions;
import com.ibm.mq.MQQueue;
import com.ibm.mq.MQQueueManager;

public class SendMessage {
	
     //定义队列管理器和队列的名称
     private static String qmName; 
     private static String qName;
     private static MQQueueManager qMgr;
     
     static{
    	 //设置环境:
    	 //MQEnvironment中包含控制MQQueueManager对象中的环境的构成的静态变量,MQEnvironment的值的设定会在MQQueueManager的构造函数加载的时候起作用,
    	 //因此必须在建立MQQueueManager对象之前设定MQEnvironment中的值.
    	 MQEnvironment.hostname="bijian-PC";         	//MQ服务器的IP地址      
    	 MQEnvironment.channel="DC.SVRCONN";  			//服务器连接的通道
    	 MQEnvironment.CCSID=1381;              		//服务器MQ服务使用的编码1381代表GBK、1208代表UTF(Coded Character Set Identifier:CCSID)
    	 MQEnvironment.port=1415;						//MQ端口
    	 qmName = "QM_ORANGE";							//MQ的队列管理器名称
         qName = "Q1";								//MQ远程队列的名称
         try {
        	 //定义并初始化队列管理器对象并连接 
	    	 //MQQueueManager可以被多线程共享,但是从MQ获取信息的时候是同步的,任何时候只有一个线程可以和MQ通信。
			qMgr = new MQQueueManager(qmName);
		} catch (MQException e) {
			// TODO Auto-generated catch block
			System.out.println("初使化MQ出错");
			e.printStackTrace();
		} 
     }
     
     /**
      * 往MQ发送消息
      * @param message
      * @return
      */
     public static int sendMessage(String message){
    	 int result=0;
    	 try{	
    		 //设置将要连接的队列属性
    	     // Note. The MQC interface defines all the constants used by the WebSphere MQ Java programming interface 
    	     //(except for completion code constants and error code constants).
    	     //MQOO_INPUT_AS_Q_DEF:Open the queue to get messages using the queue-defined default.
    	     //MQOO_OUTPUT:Open the queue to put messages.
    		 /*目标为远程队列,所有这里不可以用MQOO_INPUT_AS_Q_DEF属性*/
    		 //int openOptions = MQC.MQOO_INPUT_AS_Q_DEF | MQC.MQOO_OUTPUT;
    		 /*以下选项可适合远程队列与本地队列*/
    		 int openOptions = MQC.MQOO_OUTPUT | MQC.MQOO_FAIL_IF_QUIESCING;
	         //连接队列 
	         //MQQueue provides inquire, set, put and get operations for WebSphere MQ queues. 
	         //The inquire and set capabilities are inherited from MQManagedObject. 
    		 /*关闭了就重新打开*/
 			if(qMgr==null || !qMgr.isConnected()){
 				qMgr = new MQQueueManager(qmName);
 			}
	         MQQueue queue = qMgr.accessQueue(qName, openOptions); 	        
	         //定义一个简单的消息
	         MQMessage putMessage = new MQMessage(); 
	         //将数据放入消息缓冲区
	         putMessage.writeUTF(message);	
	         //设置写入消息的属性(默认属性)
	         MQPutMessageOptions pmo = new MQPutMessageOptions();	        
	         //将消息写入队列 
	         queue.put(putMessage,pmo); 
	         queue.close();
    	 }catch (MQException ex) { 
             System.out.println("A WebSphere MQ error occurred : Completion code " 
             + ex.completionCode + " Reason code " + ex.reasonCode); 
             ex.printStackTrace();
         }catch (IOException ex) { 
             System.out.println("An error occurred whilst writing to the message buffer: " + ex); 
         }catch(Exception ex){
             ex.printStackTrace();
         }finally{
        	 try {
 				qMgr.disconnect();
 			} catch (MQException e) {
 				e.printStackTrace();
 			}
          }
    	 return result;
     }
     
     public static void main(String args[]) {
    	 sendMessage("this is a test");
     }
}

        运行结果,java客户端将消息发送到QM_ORANGE队列管理器的远程Q1队列,QM_ORANGE队列管理器又将消息发送到QM_APPLE队列管理器的本地Q1队列。


GetMessage.java

package com.bijian.study;

import java.io.IOException;
import com.ibm.mq.MQC;
import com.ibm.mq.MQEnvironment;
import com.ibm.mq.MQException;
import com.ibm.mq.MQGetMessageOptions;
import com.ibm.mq.MQMessage;
import com.ibm.mq.MQQueue;
import com.ibm.mq.MQQueueManager;

public class GetMessage {
	
     //定义队列管理器和队列的名称
     private static String qmName; 
     private static String qName;
     private static MQQueueManager qMgr;
     
     static{
    	 //设置环境:
    	 //MQEnvironment中包含控制MQQueueManager对象中的环境的构成的静态变量,MQEnvironment的值的设定会在MQQueueManager的构造函数加载的时候起作用,
    	 //因此必须在建立MQQueueManager对象之前设定MQEnvironment中的值.
    	 MQEnvironment.hostname="bijian-PC";         	//MQ服务器的IP地址      
    	 MQEnvironment.channel="DC.SVRCONN";  			//服务器连接的通道
    	 MQEnvironment.CCSID=1381;              		//服务器MQ服务使用的编码1381代表GBK、1208代表UTF(Coded Character Set Identifier:CCSID)
    	 MQEnvironment.port=1416;						//MQ端口
    	 qmName = "QM_APPLE";							//MQ的队列管理器名称
         qName = "Q1";								//MQ远程队列的名称
         try {
        	 //定义并初始化队列管理器对象并连接 
	    	 //MQQueueManager可以被多线程共享,但是从MQ获取信息的时候是同步的,任何时候只有一个线程可以和MQ通信。
			qMgr = new MQQueueManager(qmName);
		} catch (MQException e) {
			// TODO Auto-generated catch block
			System.out.println("初使化MQ出错");
			e.printStackTrace();
		} 
     }
     
     /**
      * 从队列中去获取消息,如果队列中没有消息,就会发生异常,不过没有关系,有TRY...CATCH,如果是第三方程序调用方法,如果无返回则说明无消息
      * 第三方可以将该方法放于一个无限循环的while(true){...}之中,不需要设置等待,因为在该方法内部在没有消息的时候会自动等待。
      * @return
      */
     public static String getMessage() {
    	 String message=null;
    	 try{    		 
    		 //设置将要连接的队列属性
    	     // Note. The MQC interface defines all the constants used by the WebSphere MQ Java programming interface 
    	     //(except for completion code constants and error code constants).
    	     //MQOO_INPUT_AS_Q_DEF:Open the queue to get messages using the queue-defined default.
    	     //MQOO_OUTPUT:Open the queue to put messages.
    		 int openOptions = MQC.MQOO_INPUT_AS_Q_DEF | MQC.MQOO_OUTPUT;
	    	 MQMessage retrieve = new MQMessage();
	    	 //设置取出消息的属性(默认属性)
	    	 //Set the put message options.(设置放置消息选项) 
    		 MQGetMessageOptions gmo = new MQGetMessageOptions(); 
    		 gmo.options = gmo.options + MQC.MQGMO_SYNCPOINT;//Get messages under sync point control(在同步点控制下获取消息) 
    		 gmo.options = gmo.options + MQC.MQGMO_WAIT;  // Wait if no messages on the Queue(如果在队列上没有消息则等待) 
    		 gmo.options = gmo.options + MQC.MQGMO_FAIL_IF_QUIESCING;// Fail if Qeue Manager Quiescing(如果队列管理器停顿则失败) 
    		 gmo.waitInterval = 1000 ;  // Sets the time limit for the wait.(设置等待的毫秒时间限制) 
    		 /*关闭了就重新打开*/
 			if(qMgr==null || !qMgr.isConnected()){
 				qMgr = new MQQueueManager(qmName);
 			}
	         MQQueue queue = qMgr.accessQueue(qName, openOptions); 
	         // 从队列中取出消息
	         queue.get(retrieve, gmo);
	         message = retrieve.readUTF();	
	         System.out.println("The message is: " + message); 
	         queue.close();
    	 }catch (MQException ex) { 
             System.out.println("A WebSphere MQ error occurred : Completion code " 
             + ex.completionCode + " Reason code " + ex.reasonCode); 
         }catch (IOException ex) { 
             System.out.println("An error occurred whilst writing to the message buffer: " + ex); 
         }catch(Exception ex){
             ex.printStackTrace();
         }finally{
        	 try {
				qMgr.disconnect();
			} catch (MQException e) {
				e.printStackTrace();
			}
         }
    	 return message;
     }
     
     public static void main(String args[]) {
    	 getMessage();
     }
}

        运行结果,控制台输出

The message is: this is a test

        此时,QM_APPLE队列管理器中的Q1队列中的消息已被取出。


 

参考文章: http://blog.csdn.net/fenglibing/article/details/4161441





已有 0 人发表留言,猛击->> 这里<<-参与讨论


ITeye推荐



相关 [mq 通讯 完整] 推荐:

与MQ通讯的完整JAVA程序

- - 编程语言 - ITeye博客
        本文实例是基于 WebSphere MQ中将消息发送至远程队列的配置的基础上的,且如果要能正常运行并发送、接收消息,还需要在两个队列管理器(QM_ORANGE和QM_APPLE)上做如下配置或修改.         1.创建名称为DC.SVRCONN的服务器连接通道.         2.将队列管理器的通道认证记录设置为“已禁用”.

闲扯kafka mq

- - 开源软件 - ITeye博客
本文主要讲解关于kafka mq的设计思想及个人理解. 关于kafka的详细信息,大家可以参考官网的文献 http://kafka.apache.org/documentation.html这是一篇相当不错的文章,值得仔细研读. 第一个问题:消息队列(Message Queue)是干嘛用的. 首先,要对消息队列有一个基本的理解.

[MQ]关于ActiveMQ的配置

- - 企业架构 - ITeye博客
  目前常用的消息队列组建无非就是MSMQ和ActiveMQ,至于他们的异同,这里不想做过多的比较. 简单来说,MSMQ内置于微软操作系统之中,在部署上包含一个隐性条件:Server需要是微软操作系统. (对于这点我并去调研过MSMQ是否可以部署在非微软系统,比如:Linux,只是拍脑袋想了想,感觉上是不可以).

在Tomcat 6.0下用JNDI连接IBM MQ 6.0的配置方法

- - 行业应用 - ITeye博客
假设在IBM MQ中定义的队列管理器的名为QueueManager, 端口1414,CCSID 437 ,创建名为LQ1,LQ2的队列分别用于发送和接收消息, 服务器连接通道名为SVRCONN. 确保在项目的Classpath中导入了以下的jar包:. 如果需使用spring的JmsTemplate方式来读写MQ,还需要导入.

消息队列mq的3个使用场景

- - 互联网 - ITeye博客
2、tomcat controller取到请求后向rocketmq 发送一个msg,将msg id返回给app,同时在redis里缓存msg状态为init(设置定时时间,时间到后清除). 3、client(app/h5/小程序) 通过msg id,定时向server获取msg处理状态. (init时 画圈,没有时返回繁忙,fail时返回处理失败).

GitHub - 事务最终一致性 - MQ保证消息一致性

- -
使用rabbitMq来简单实现分布式事务的最终一致性 版本如下:. 目录 名称 访问地址. gateway 网关+路由. rabbitMq -> 注册中心 -> 网关 -> 订单 -> 支付. 3、发送消息(采用确认模式). 4、mq收到消息之后给生产端一个确认消息. 5、生产端监听这个确认消息.

使用MQ的时候,怎么确保消息100%不丢失?

- - 掘金后端本月最热
面试官在面试候选人时,如果发现候选人的简历中写了在项目中使用了 MQ 技术(如 Kafka、RabbitMQ、RocketMQ),基本都会抛出一个问题: 在使用 MQ 的时候,怎么确保消息 100% 不丢失. 这个问题在实际工作中很常见,既能考察候选者对于 MQ 中间件技术的掌握程度,又能很好地区分候选人的能力水平.

关于聚合数据窃取用户通讯录的完整分析

- - FreeBuf.COM
iOS SDK Demo  的时候,发现会请求通讯录权限 … 当时刚睡醒没多久,拿了自己在用的手机,还点了 Allow. 然后我就在 Charles 里发现了一个很大的请求,很好奇,便对 SDK 进行了逆向工程,拿到了解密方法,发现我的整个通讯录都被上传了. 于是我在 V2EX 发了一个帖子  请不要使用聚合数据的 SDK,后来被聚合的工作人员看到,他们是 这样回复的:.

不恰当使用线程池处理 MQ 消息引起的故障

- - 码蜂笔记
业务部门反应网站访问特别慢,负责运维监控的同事说MQ消息队列积压了,中间件的说应用服务器内存占用很高,GC 一直回收不了内存,GC 线程占了近 100% 的 CPU,其他的基本上都在等待,数据库很正常,完全没压力. 没啥办法,线程、堆 dump 出来后,重启吧,然后应用又正常了. 这种故障之前其实也碰到过了,分析了当时 dump 出来的堆后发现,处理 MQ 消息的线程池的队列长度达百万级别,占用了超过 1.3G 内存,这些内存都是没法回收的.

完整

- None - 韩寒
在年29的白天,钱云会的手表视频被曝光了. 很巧,我也有一块和钱村长一样的手表,是我父亲在我今年生日的时候在上海国际赛车场里送给我的. 但是我没有怎么用过它,所以和我约会的姑娘大可放心. 我有一个朋友质疑说,为什么视频里先出现了老钱的脸,是否太欲盖弥彰了. 这个我倒是可以来解释一下,因为我这款手表的摄像头在表盘的十二点钟位置,而开摄像头的按键在右边的表盘侧面上,按压两秒就打开摄像头,这个时候摄像头的小灯会发出蓝光,5秒钟以后熄灭,表示录制开始.