IBM Websphere MQ 使用详解
1.MQ安装配置
安装和配置的内容,可参考https://jingyan.baidu.com/article/2fb0ba409f4e1e00f2ec5fd3.html
下面是安装配置后的截图:
2.队列和通道的理解
队列对应Buffer,通道对应Channel。
根据通道是本方还是对方,启用相应的通道(即服务)。
远程队列_remote发送到对方的本地队列_local。
比如,在abc_remote上放入消息“123”,在cz_local队列上就能获取“123”。
更详细的理解请点击
3.java代码
3.1 阻塞模式
import grp.pt.util.StringUtil;import org.apache.log4j.Logger;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.MQPutMessageOptions;
import com.ibm.mq.MQQueue;
import com.ibm.mq.MQQueueManager;
import com.ibm.mq.MQSimpleConnectionManager;
import com.river.common.UploadFileUtil;/*** MQ配置 连接信息*/
public class MqHandler {private static Logger log = Logger.getLogger(MqHandler.class);private static String MQ_IP = UploadFileUtil.getFromPro("mqconfig", "MQ_IP");private static int MQ_PORT = Integer.parseInt(UploadFileUtil.getFromPro("mqconfig", "MQ_PORT"));// 队列管理器private static String MQ_QMANAGER = UploadFileUtil.getFromPro("mqconfig","MQ_QMANAGER");// mq服务方通道private static String MQ_CHANNEL = UploadFileUtil.getFromPro("mqconfig","MQ_CHANNEL");// mq通讯编码集private static int MQ_CCSID = Integer.parseInt(UploadFileUtil.getFromPro("mqconfig", "MQ_CCSID"));// mq请求方发送队列private static String MQ_CLENT_SENDQUEUE = UploadFileUtil.getFromPro("mqconfig", "MQ_CLENT_SENDQUEUE");// mq请求方接收队列private static String MQ_CLENT_RECQUEUE = UploadFileUtil.getFromPro("mqconfig", "MQ_CLENT_RECQUEUE");// MQ自带连接池private static MQSimpleConnectionManager myConnMan = null;// 异步长连接接收队列管理器private MQQueueManager queueManager = null;// 异步发送队列private MQQueue SendQueue = null;// 异步接收队列public MQQueue ReceiveQueue = null;/*** 构造方法*/public MqHandler() throws Exception {init();}/*** 异步发送消息* * @param msg* @param messageId* @param charSet* @throws Exception*/public void sendMsg(String msg, String messageId, String charSet)throws Exception {try {if (StringUtil.isEmpty(charSet)) {charSet = "GBK";}byte[] msgId = StringUtil.isEmpty(messageId) ? null : messageId.getBytes(charSet);// 打开队列queueManager = new MQQueueManager(MQ_QMANAGER, myConnMan);SendQueue = queueManager.accessQueue(MQ_CLENT_SENDQUEUE,MQC.MQOO_OUTPUT | MQC.MQPMO_NEW_MSG_ID| MQC.MQOO_FAIL_IF_QUIESCING, null, null, null);// 初始化消息选项MQPutMessageOptions pmo = new MQPutMessageOptions();// 确保每次发送前为消息自动生成唯一的msgIdpmo.options = pmo.options + MQC.MQPMO_NEW_MSG_ID;// 如果设置了该参数,则发送后必须调用commit功能。否则无法将消息发送出pmo.options = pmo.options + MQC.MQPMO_SYNCPOINT;// 创建消息对象MQMessage outMsg = new MQMessage();// 设置MQMD格式字段outMsg.format = MQC.MQFMT_STRING;outMsg.messageId = msgId == null ? MQC.MQMI_NONE : msgId;outMsg.encoding = MQ_CCSID;outMsg.characterSet = MQ_CCSID;// 消息发送时必须以字节流的方式发送outMsg.write(msg.getBytes(charSet));// 在队列上放置消息SendQueue.put(outMsg, pmo);// 和MQC.MQPMO_SYNCPOINT属性对应。如果设置了该属性,则发送后需要提交。queueManager.commit();} finally {close(SendQueue);}}/*** 接收消息* @param correlationId* @param charSet* @return* @throws Exception*/public String receive(String correlationId, String charSet)throws Exception {try {if (StringUtil.isEmpty(charSet)) {charSet = "GBK";}queueManager = new MQQueueManager(MQ_QMANAGER, myConnMan);ReceiveQueue = queueManager.accessQueue(MQ_CLENT_RECQUEUE,MQC.MQOO_INPUT_AS_Q_DEF | MQC.MQOO_INQUIRE| MQC.MQOO_FAIL_IF_QUIESCING);MQMessage respMessage = new MQMessage();// 设置MQMD 格式字段respMessage.format = MQC.MQFMT_STRING;// 设置编码格式与MQ服务一致respMessage.encoding = MQ_CCSID;// 设置字符集与MQ服务一致respMessage.characterSet = MQ_CCSID;MQGetMessageOptions gmo = new MQGetMessageOptions();gmo.options = gmo.options + MQC.MQGMO_WAIT; // 如果设置了该参数,则当前线程将阻塞,直到等到回复的消息或超时gmo.waitInterval = MQC.MQWI_UNLIMITED;if (StringUtil.isNotEmpty(correlationId)) {// 如果设置了该参数,则根据消息的correlId去匹配对应的响应消息gmo.matchOptions = MQC.MQMO_MATCH_CORREL_ID;respMessage.correlationId = correlationId.getBytes(charSet);}ReceiveQueue.get(respMessage, gmo);byte[] msgBuffer = new byte[respMessage.getMessageLength()];respMessage.readFully(msgBuffer);String respMsg = new String(msgBuffer, charSet);return respMsg;} finally {close(ReceiveQueue);}}/*** 初始化连接* * @param type* @throws MQException*/private void init() throws MQException {MQEnvironment.hostname = MQ_IP;MQEnvironment.port = MQ_PORT;MQEnvironment.CCSID = MQ_CCSID;MQEnvironment.channel = MQ_CHANNEL;MQEnvironment.properties.put(MQC.TRANSPORT_PROPERTY,MQC.TRANSPORT_MQSERIES_CLIENT);myConnMan = new MQSimpleConnectionManager();myConnMan.setActive(MQSimpleConnectionManager.MODE_AUTO);myConnMan.setTimeout(3600000);myConnMan.setMaxConnections(75);myConnMan.setMaxUnusedConnections(50);MQEnvironment.setDefaultConnectionManager(myConnMan);log.info("初始化队列管理器receiverQueueManager....." + MQ_QMANAGER);}private void close(MQQueue queue) {try {if (queue != null && queue.isOpen) {queue.close();}if (queueManager != null) {queueManager.disconnect();}} catch (MQException ex) {log.error("", ex);}}
}
网上类似的代码很多,下面对代码中的细节进行解释。
初始化
MQEnvironment的静态属性:hostname、port、CCSID等初始化后,MQQueueManager利用这些属性来构造实例对象。
MQSimpleConnectionManager是简单的连接池,当设置了连接池后MQEnvironment.setDefaultConnectionManager(myConnMan),再使用close方法时,并不会破坏连接,而是归还池中。
连接池详细内容可以参考官方的API
消息参数
- MQOO_FAIL_IF_QUIESCING
如果队列管理器停顿就取不到消息 - MQOO_INPUT_AS_Q_DEF
读取后消息从队列中移除,与之相对的是MQOO_BROWSE,读取后消息还在。 - MQGMO_WAIT
当队列中没有消息时,等待有消息时才返回。相当于阻塞IO。 - MQWI_UNLIMITED
无限等待时间,必须设置了waitInterval.xxx并和上条属性结合才能起到阻塞的效果。这个等待时间可为任意值。 - MQOO_INQUIRE
如果需要获取队列深度等信息的时候,就要加上此参数。 - MQPMO_NEW_MSG_ID
确保每次发送前为消息自动生成唯一的msgId。 - MQPMO_SYNCPOINT
如果设置了该参数,则发送后必须调用commit功能。否则无法将消息发送出。queueManager.commit()。
更多参数参见官方说明
3.2测试类
import org.jfree.util.Log;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;import com.ctjsoft.treasury.util.MqHandler;@RunWith(JUnit4.class)
public class MqTest {@Testpublic void mqSend(){try {MqHandler mq = new MqHandler();mq.sendMsg("sq--send1", null, "gbk");mq.sendMsg("sq--send2", null, "gbk");mq.sendMsg("sq--send3", null, "gbk");} catch (Exception e) {// TODO Auto-generated catch blocke.printStackTrace();}}@Testpublic void mqReceive(){try {MqHandler mq = new MqHandler();String msg = mq.receive(null, "gbk");System.out.println(msg);String msg1 = mq.receive(null, "gbk");System.out.println(msg1);String msg2 = mq.receive(null, "gbk");System.out.println(msg2);System.out.println("任务完成");} catch (Exception e) {Log.error("", e);}}}
测试过程中出现的问题举例:
- MQJE001: 完成代码是 2,原因为 2033
当前消息队列为空,如果不是阻塞模式的获取消息,队列为空就会抛出此异常。 - MQJE001: MQException 出现:完成代码是 2,原因为 2018
MQJI002: 未连接至队列管理器。
消息通道服务被关闭 - 原因为 2080
代表消息的实际长度超过了你在程序代码中设置的缓冲区长度
感觉有用请点赞
IBM Websphere MQ 使用详解相关推荐
- 配置IBM WebSphere MQ及注意事项
配置IBM WebSphere MQ及注意事项 安装IBM WebSphere MQ 搭建测试环境 遇到的问题 配置IBM WebSphere MQ及注意事项 在项目中用到IBM WebSphere ...
- 使用c#操作IBM WebSphere MQ
IBM WebSphere MQ 5.3升级到CSD05之后,提供了使用.net操作MQ的类库,安装完成之后,会在MQ的安装目录的bin文件夹下面多出一个amqmdnet.dll文件,把这个DLL作为 ...
- IBM Websphere MQ 基础0:Linux下安装IBM MQ 7.5
背景 无.公司就是用的IBM WebSphere MQ作为消息中间件,本文记载how install it. 一. 安装包检查与准备 1. 检查 通过rpm –qa检查是否已经安装 IBM MQ,若已 ...
- ibm服务器安装aix系统,在aix下安装ibm websphere MQ 7.0
最近为项目设置灾难恢复环境(包括和应用程序的灾难恢复). Oracle同步已通过DATAGUARD实现. 现在要将应用程序部署到灾难恢复室中的服务器,该应用程序软件使用ibm Websphere. M ...
- IBM WebSphere MQ安装及使用教程
IBM WebSphere MQ安装及使用教程 Linux下websphere MQ安装 #软件安装目录 mkdir /opt/mqm #添加用户组mqm groupadd mqm useradd - ...
- IBM WebSphere MQ 7.5基本用法
一.下载7.5 Trial版本 http://www.ibm.com/developerworks/downloads/ws/wmq/ 这是下载网址,下载前先必须注册IBM ID,下载完成后一路Nex ...
- Hyperic HQ监控IBM WebSphere MQ
2019独角兽企业重金招聘Python工程师标准>>> IBM公司的WebSphere MQ(MQSeries的前身)是一个信息服务器,使您可以轻松地在不同平台交流信息,整合新的和现 ...
- IBM WebSphere MQ 常见报错码及解决方案
在IBM WebSphere MQ的使用过程中会遇见的常见报错描述和初步解决方案,希望能帮助各位解决使用中可能存在的问题. 序号 错误编号1 原因 1 2033 可能由于队列为空,取不到值引起的 2 ...
- spring boot整合IBM WebSphere MQ,并配置多个队列管理器
IBM WebSphere MQ概述 网上关于IBM WebSphere MQ的资料挺少的,毕竟是一项老技术,整理一下从零开始对于IBM WebSphere MQ的理解 IBM WebSphere M ...
最新文章
- View的Touch事件分发(一.初步了解)
- SAP MM 事务代码MI31之思考之续集
- 解决 FtpClient 类无法导入 .
- Linux安装ImageMagick与JMagick完成过程及配置
- 将数组作为参数,调用该函数时候给的是数组地址还是整个数组
- FFMPEG解码多线程
- ASP程序密码验证漏洞
- windows 互斥量内核对象 Mutex
- 【钉钉机器人 + 爬虫 + celery】定时发送微博热搜 + 定时发布财经新闻
- python实现邻接矩阵转邻接表
- 港科报道 | 8位校友入选香港25青年科创先锋人物
- [re入门]一个简单的加密程序的逆向破解与解密
- 计算机网络地址块例题,计算机网络习题计算机络习题.ppt
- Deep Crossing: Web-Scale Modeling without Manually Crafted Combinatorial Features(2016)
- git pull报错Pulling is not possible because you have unmerged files
- ubuntu 生成桌面快捷方式(有图)
- 为改善Siri功能Apple收购语音助理初创公司
- Win10无法更改账户名称怎么办
- 计算机与医药信息学,浙江大学药物信息学研究所
- CSP202112-3登机牌条码