IBM MQ简单开发和应用

末蓝、 2022-06-10 08:59 1064阅读 0赞

IBM MQ经常被一些政府公共部门，银行等企业用来做数据传输和报文收发，在互联网应用的开发中较少见到，属于一种比较老旧的应用。这里以IBM Websphere MQ 7.5版本为例介绍一下MQ的的简单开发和应用。

首先到IBM官网下载目前最新的7.5版本，https://www.ibm.com/developerworks/cn/downloads/ws/wmq/

有90天试用版和Advanced版可供下载，其中Advanced版本提供了全部功能，没有日期限制，只是减少了一些后端支持，所以推荐下载。

首次安装会问你网络中有没有win2000及一上版本的电脑，我们选没有就可以了。本人亲测在winXP，win7，win10上均可正常安装使用。

安装成功后，首先需要了解一些概念：

1、队列管理器

队列管理器是MQ的主要部件，一个队列管理器监听一个端口，里面可以配置监听器，队列，通道，触发器等等功能。在使用JAVA程序与MQ进行通信时，要使用到队列管理器的名字。

一般搭建MQ服务器的第一件事就是新建队列管理器，并指定其接口

2、队列

队列，顾名思义就是一个消息的集合，可以理解成一个先入先出的栈，或者一个List。队列里面存储的就是收到的报文或者将要发出的报文。如果把队列管理器理解成一个数据库，那么队列就好比一张表。但是这个表是有顺序的，在JAVA操作中可以看成一个先入先出的栈结构。MQ客户端发出消息其实就是把消息压入栈中，而收取消息就是把消息从栈中弹出来。

2.1、本地队列

故名思议，该队列是存储在本地的，占用当前服务器的内存资源，一条消息进入本地队列就代表当前服务器收到了消息。可以在JAVA代码中很轻松的把数据放入本地队列或者从本地队列取出，从而实现了数据收发的工作。本地队列建立比较容易，点两下鼠标就能建好

![Center][]

2.2、传输队列

传输队列是一种特殊的本地队列，新建本地队列时，默认是普通队列，如果要建立传输队列需要修改【使用情况】为“传输”，如下图

![Center 1][]

传输队列的作用非常重要，传输队列是连接本地队列，通道，远程队列的桥梁。当我们把数据放到远程队列上时，远程队列会把数据放置到传输队列上，然后经过发送通道发出，直至另一台服务器收取。

**2.3、远程队列**

远程队列相当于把别的服务器上的本地队列映射到当前服务器上，当你把一条信息放到当前服务器远程队列上，那么你就可以再另一台服务器上的本地队列上看见这条消息。换句话说，“发送”这个行为就变成了把信息放到远程队列上的动作

远程队列有这样几个要素必须要填写

![Center 2][]

1、远程队列名称：本地的远程队列名称，随便填就可以

2、远程队列：接收消息那台服务器的本地队列名称，必须与那台服务器一致，不能乱填。

3、远程队列管理器：接收消息那台服务器的队列管理器的名称，必须一致，不能乱填。

4、传输队列：本地建立的传输队列。

3、通道

通道就是连接两台服务器，或者客户机和服务器，进行数据IO实际操作的组件。换句话说，只要有信息通过MQ收发，就必须有通道。

3.1、服务器连接通道

该通道就是赋予JAVA客户端或者其他设备与MQ服务交互信息的通道，只有建立了这个通道，我们的JAVA程序才能对接向MQ服务器发送或者提取消息。有了这个通道之后，我们的JAVA程序可以在没安装MQ软件的机器上从有MQ的服务器上收发数据。这样就不必两个设备都安装MQ软件了。该通道建立非常容易，全部默认一键就好了。

3.2、发送方通道

该通道就是控制数据从本地向远程发送的通道，这个通道必须填写如下几个要素

![Center 3][]

通道名称：可以随便填

连接名称：接收消息服务器的IP地址和端口，端口要写在括号内

传输队列：填写之前建立好的本地的传输队列的名称。

发送通道用在Sender--Receiver这种应用模式中，在发送端的服务器需要建立【本地队列】【传输队列】【远程队列】和【发送方通道】，在接收消息的服务器要建立【本地队列】和【接收方通道】即可，拓扑图如下

注意：本机【发送方通道】的名称要与另外一台服务器【接收方通道】的名称一致；

Sender-Receiver模式或者Sender-Requestor模式都会用到【发送方通道】

![Center 4][]

3.3、接收方通道

该通道就是时刻准备接收其他机器发送通道发送来数据的通道，建立【接收方通道】非常容易，只需要指定一下协议就可以了，默认TCP，一般一键完成。

注意:接收方通道的名称要与另外那台机器上发【送方通道】的名称一致

3.4、服务器通道

服务器通道用法和【发送方通道】基本一样，都需要填写IP地址和【传输队列】的名字，用在Server-Receiver这种模式中可以完全取代【发送方通道】，就连拓扑图也只是把名字改了。一般配置如下

![Center 5][]

【发送方通道】和【服务器通道】的区别是：

(1)连接名称这里的IP地址不是必填了，当不填的的时候，使用的是Server-Requester方式通信，可以实现消息一对多的广播发送，而填IP的时候Server-Receiver模式下是发送方主动把消息塞到接收方的本地队列里，用于一对一的通信；

(2)【服务器通道】支持Server-Requester模式通信而【发送方通道】不支持

(3)在Server-Requester模式下可以实现一对多的发送，而Sender--Receiver只能进行一对一的发送

和【发送方通道】的相同点是：

(1)都需要填写传输队列的名字；

(2)Server-Receiver模式下（填IP）本机的【服务器通道】名称要与远程服务器【接收方通道】的名称一致；

3.5、请求方通道

【请求方通道】一般用在Server-Requester模式下，与【服务器通道】相对应，消息发送方服务器的【服务器通道】名称必须与消息接受服务器【请求方通道】一致。消息接受服务器有了【请求方通道】之后就不需要【接收方通道】了，其拓扑图如下

![Center 6][]

Server-Requester模式和Server-Receiver/Sender--Receiver模式的相同点是：消息发送服务器都是把消息放到远程队列里面去发送给消息接收服务器

不同点是：（1）消息发送服务器无需配置消息接收服务器的IP地址，可以实现广播，也就是一对多的消息发送。而其他两种模式只能实现一对一的发送

(2)在接收消息的服务器上必须配置【请求方通道】并指定发送方服务器的IP地址。

(3)在接收消息的服务器上无需配置【接收方通道】

另外还有一种模式Sender-Requester，用于实现一对一的通信，实现起来比一上三种都要复杂，需要消息发送服务器和消息接收服务器都配置上互相的地址，增强了安全性。

在发送服务器上，和其他三种方式一样，需要【本地队列】【传输队列】【远程队列】，这三个队列配置方法于前三种一模一样；需要一个通道【发送方通道】，并指定接受服务器的IP地址

在接收消息的服务器上，与其他三种方式一样需要【本地队列】【传输队列】【远程队列】，只需要一个通道就是【请求方通道】，并配置上发送服务器的IP地址，拓扑图如下。

![Center 7][]

下面再把四种通讯模式梳理一下

相同点，都需要建立【本地队列】【传输队列】【远程队列】，而且建立的方式的参数完全一样，只是通道配置不一样。

Sender--Receiver：用于一对一的通讯，发送方服务器必须指定接收服务器的地址，而接收服务器配置非常简单，只需要一个【接收方通道】就行了，可以接收很多发送服务器发来的消息，相当于多对一。

发送服务器配置【发送方通道】并指明接收服务器的IP

接受服务器配置【接收方通道】，全部默认，两个通道的名称必须一致

Server-Receiver：效果与Sender--Receiver一模一样，进行一对一的通讯。

发送服务器配置【服务器通道】并指明接收服务器的IP

接受服务器配置【接收方通道】，全部默认，两个通道的名称必须一致

Server-Requester：用于实现一对多的通讯

发送服务器配置【服务器通道】，不指明IP

接受服务器配置【请求方通道】，指明发送服务器的IP，两个通道的名称必须一致

Sender-Requester：用于实现严格的一对一通讯

发送服务器配置【服务器通道】，并指明接收服务器的IP

接受服务器配置【请求方通道】，指明发送服务器的IP，两个通道的名称必须一致

应用场景

1、网络中只有一台装有MQ的服务器，一个或多个JAVA客户端与这台服务器收发报文

这种情况非常简单，只需建立一个【队列管理器】，然后在这个队列管理器下面建立一个【本地队列】和一个【服务器连接通道】就可以了。

具体的方法可以参考这一篇文章及其提供的源码：http://blog.csdn.net/qq\_17616169/article/details/54633005

2、网络中有两台服务器都安装了MQ，两台服务器之间进行相互的报文收发

这种情况需要用到多个队列和多个通道，最简单的实施方案Sender--Receiver模式如下：

A服务器：

（1）建立一个【队列管理器】

（2）建立一个【本地队列】，一个【传输队列】，一个【远程队列】

（3）建立一个【发送方通道】，一个【接收方通道】

（4）如果需要和JAVA客户端通信，还需要一个【服务器连接通道】

B服务器：与A服务器一样，只是名字不一样，【发送方通道】的配置不一样

可以参考这一篇文章和提供的源码：http://blog.csdn.net/xpsharp/article/details/32100977

我就是按照上面这一片文章为基础做的。

下面主要讲一下这种模式的配置细节，首先是三个队列的设置

![Center 8][]

其中，LQ\_00000000是A机器的【本地队列】，LQ\_88888888是B机器的【本地队列】，这两个队列通过A机器的【远程队列】关联起来。另外QM\_88888888是B服务器的【队列管理器】，也要配置到A机器的【远程队列】上。

下面是A机器的三个通道的设置

![Center 9][]

其中36.32.160.169是B机器的IP地址，1414为B机器MQ的监听端口，88888888.00000000是B机器的【发送方通道】的名称。

DC.SVRCONN是用来给JAVA程序接入用的【服务器连接通道】

然后来看一下B服务器的三个队列的设置：

![Center 10][]

B服务器上的【远程队列】的远程队列LQ\_00000000就是A服务器的本地队列，远程队列管理器就是A服务器的队列管理器，队列名称可以随意填，传输队列为服务器B的【传输队列】

还有三个通道如下

![Center 11][]

【发送方通道】的连接名称应该填A服务器的IP地址，因为这样才可以互通，传输队列填服务器B的传输队列。

JAVA代码的操作

java代码上其实不难写，只是容易出现各种奇奇怪怪的报错。首先要把开发的环境搭建好，最重要的一点就是要引入完整的jar包，下面这9个jar包不可少

![Center 12][]

注意在maven的官网上，虽然有connector-1.0的页面，但是下载连接已经失效了，也就是说使用maven自动下载jar包是不可以的，必须手工导入。

导入后我们一般都会做一件事就是向MQ服务器发送消息，在发送消息之前需要配置一些项目。

MQQueueManager是一个很重要的类，它就相当与MQ里面的【队列管理器】。所以在new它的时候肯定要传入队列管理器的名字。有两种构造方法，分别是

new MQQueueManager("本地的【队列管理器】的名字")；
    或者 new MQQueueManager("本地的【队列管理器】的名字"，属性Hashtable)；

第一种方式也需要属性，不过属性是在一个全局静态变量中配置的，如下

MQEnvironment.hostname = "127.0.0.1";//这个是MQ的服务器的IP地址
    MQEnvironment.channel = "SERVERCONN";//这个是MQ服务器的通道名，对于JAVA一般是【服务器连接通道】
    MQEnvironment.port = 1414;//这个是MQ服务器监听的端口
    MQEnvironment.CCSID = 1381;//这个是编码格式的代号,UTF-8为1381
    MQEnvironment.userID = "MUSR_MQADMIN";//这个是登录MQ服务器用的用户名，需要在MQ服务器上设置
    MQEnvironment.password = "123456";//这个是登录MQ服务器用的密码

如果是使用第二种构造函数，那么也大同小异

Hashtable properties = new Hashtable();
    properties.put("hostname", "127.0.0.1");
    properties.put("port", new Integer(1414));
    properties.put("channel", "DC.SVRCONN");
    properties.put("CCSID", new Integer(1381));
    properties.put("userID","MUSR_MQADMIN");
    properties.put("password","123456");

然后把这个Hashtable传入到构造函数中就可以获得一个MQQueueManager对象了，我们后面需要用这个对象来获得MQQueue对象。

MQQueue这个类代表的就是一个队列，操作它和操作MQ上的队列是差不多的。这个类能够查看队列的深度，向队列里放入消息或者取出消息，非常重要。

MQQueue queue = qMgr.accessQueue("队列名称", int模式, null, null, null);

其中队列名称可以是【本地队列】的名字也可以是【远程队列】的名字，如果你想把消息放到当前连接的MQ服务器上，那么就填【本地队列】的名字，如果你想让当前MQ把消息转发到另外一台MQ服务器上，那么就填远程队列的名字。

第二个参数，一般第一个参数是【本地队列】的时候就填49，【远程队列】就填16

这是非常重要的一步，只有成功建立JAVA与MQ服务器队列的连接，我们才能放入数据和取出数据。JAVA不需要考虑通道是怎么设置的，也不需要管理消息是如何转发的。JAVA只负责把消息放到队列或者从队列取出，其他的工作都是由MQ中队列和响应通道的配置完成的。

但是这一步稍有不慎就会报出错误，一般有个100%会出现的错误就是2035错误码，此时MQ服务器正常启动，并且在端口上监听好，但是你的JAVA没有权限去连接。出现2035只需要在命令行里把MQ的用户校验关掉就好了。

其他的错误要么是网络不通，要么是队列或者通道配置错了，随机应变吧。

下面是消息放入队列（发送消息）和从队列里取出消息（接收消息）的源码

public static void sendMsg(String msgStr) {
    		int openOptions = MQC.MQOO_INPUT_AS_Q_DEF | MQC.MQOO_OUTPUT | MQC.MQOO_INQUIRE;
    		System.out.println("发送openOptions=="+openOptions);
    		MQQueue queue = null;
    		try {
    			// 建立Q1通道的连接
    			System.out.println("准备连接队列："+queueString);
    			queue = qMgr.accessQueue(queueString, openOptions, null, null, null);
    			System.out.println("队列连接成功");
    			MQMessage msg = new MQMessage();// 要写入队列的消息
    			msg.format = MQC.MQFMT_STRING;
    			msg.characterSet = 1381;
    			msg.encoding = 1381;
    			// msg.writeObject(msgStr); //将消息写入消息对象中
    			msg.writeString(msgStr);
    			MQPutMessageOptions pmo = new MQPutMessageOptions();
    			msg.expiry = -1; // 设置消息用不过期
    			queue.put(msg, pmo);// 将消息放入队列
    		} catch (Exception e) {
    			// TODO Auto-generated catch block
    			e.printStackTrace();
    		} finally {
    			if (queue != null) {
    				try {
    					queue.close();
    				} catch (MQException e) {
    					// TODO Auto-generated catch block
    					e.printStackTrace();
    				}
    			}
    		}
    	}
    
    	public static void receiveMsg() {
    		int openOptions = MQC.MQOO_INPUT_AS_Q_DEF | MQC.MQOO_OUTPUT | MQC.MQOO_INQUIRE;
    		MQQueue queue = null;
    		System.out.println("接收openOptions=="+openOptions);
    		try {
    			System.out.println("准备连接队列："+queueString);
    			queue = qMgr.accessQueue("LQ_00000000", openOptions, null, null, null);
    			System.out.println("队列连接成功,该队列当前的深度为:" + queue.getCurrentDepth());
    			System.out.println("===========================");
    			int depth = queue.getCurrentDepth();
    			// 将队列的里的消息读出来
    			while (depth-- > 0) {
    				MQMessage msg = new MQMessage();// 要读的队列的消息
    				MQGetMessageOptions gmo = new MQGetMessageOptions();
    				queue.get(msg, gmo);
    				System.out.println("消息的大小为：" + msg.getDataLength());
    				System.out.println("消息的内容：\n" + msg.readStringOfByteLength(msg.getDataLength()));
    				System.out.println("---------------------------");
    			}
    		} catch (Exception e) {
    			// TODO Auto-generated catch block
    			e.printStackTrace();
    		} finally {
    			if (queue != null) {
    				try {
    					queue.close();
    				} catch (MQException e) {
    					// TODO Auto-generated catch block
    					e.printStackTrace();
    				}
    			}
    		}
    	}

参考文章：http://www.cnblogs.com/god-S/p/4997081.html

[Center]: /images/20220610/b2e2a31bee1b40fcb854043022f8ca9c.png
[Center 1]: /images/20220610/2cd03e96ea014115abed7322c39de231.png
[Center 2]: /images/20220610/8d98b0cf44e1401287b11e3d23f98662.png
[Center 3]: /images/20220610/de9e76ca65004b49804b4237b642308a.png
[Center 4]: /images/20220610/4cc08946f6db42e0a1bb1c7dd3262341.png
[Center 5]: /images/20220610/cd3cbcabe0e142f5a34c2d3e8bf9a76e.png
[Center 6]: /images/20220610/f96e59a2a9ed4242b54e41ce098b3fcb.png
[Center 7]: /images/20220610/d33ddd823de1460fa4de9565ad18984a.png
[Center 8]: /images/20220610/2d5efa723b5940cbbd2dc2d32644115c.png
[Center 9]: /images/20220610/9040d7dfcd5948879a502a40879ecade.png
[Center 10]: /images/20220610/486ceb4008404bb583183f32485d00b3.png
[Center 11]: /images/20220610/b9a8a887dcda45ecad304e27df6e0cfd.png
[Center 12]: /images/20220610/b23ed77eff574f0382ac73adca651ac9.png