Linux上的WebSphere MQ开发快速入门

本文的目标是让您轻松实现Linux®上的IBM® WebSphere® MQ入门。由于大部分Linux服务器并没有图形用户界面，因此本文将说明如何使用命令行工具在Linux上安装和配置WebSphere MQ。本文将说明如何使用WebSphere MQ Java API创建两个示例Java™应用程序：MQSend（用于向队列发送消息）和MQGet（用于从队列接收消息）。

　　先决条件

　　WebSphere MQ V6安装包。

　　一台或两台安装了Linux的计算机。本文使用两台安装了Suse Linux Enterprise Server V9的计算机，不过也可以在一台计算机上完成所有工作。

　　Linux管理技能。

　　Java开发方面的经验将有所帮助，但并非必须的，因为我们提供了经过预编译的Java示例应用程序供下载和在命令行使用。

　　WebSphere MQ

　　WebSphere MQ 提供了连接层和应用程序的可靠集成，为SOA提供了所需的基础。WebSphere MQ的主要特色功能包括：

　　有保证的消息交付基于可靠队列的消息传递，可确保消息不会丢失或重复。基于标准的开发使用Java Message Service (JMS) API和Message Queue Interface(MQI)。 端到端安全使用安全套接字层（Secure Sockets Layer，SSL）进行身份验证、消息完整性检查和数据加密。Web服务支持用于保证Web服务传输和可靠性的简单对象访问协议（Simple Object Access Protocol，SOAP）。异步处理对应用程序进行分离，以支持进行独立于时间的处理。多平台支持支持80种以上的配置。

　　安装WebSphere MQ

　　安装WebSphere MQ的过程非常简单——直接将安装包释放到目录中即可（本文使用root主目录下的mq_install）。可以安装RPM安装包中的所有内容，但本文仅需要这部分东西。以下是WebSphere MQ V6中包括的包和目录：

以下是引用片段： IBMJava2-SDK-1.4.2-0.0.i386.rpm MQSeriesClient-6.0.0-0.i386.rpm MQSeriesConfig-6.0.0-0.i386.rpm MQSeriesFTA-6.0.0-0.i386.rpm MQSeriesIES30-6.0.0-0.i386.rpm MQSeriesJava-6.0.0-0.i386.rpm MQSeriesKeyMan-6.0.0-0.i386.rpm MQSeriesMan-6.0.0-0.i386.rpm MQSeriesMsg_Zh_CN-6.0.0-0.i386.rpm MQSeriesMsg_Zh_TW-6.0.0-0.i386.rpm MQSeriesMsg_de-6.0.0-0.i386.rpm MQSeriesMsg_es-6.0.0-0.i386.rpm MQSeriesMsg_fr-6.0.0-0.i386.rpm MQSeriesMsg_it-6.0.0-0.i386.rpm MQSeriesMsg_ja-6.0.0-0.i386.rpm MQSeriesMsg_ko-6.0.0-0.i386.rpm MQSeriesMsg_pt-6.0.0-0.i386.rpm MQSeriesRuntime-6.0.0-0.i386.rpm MQSeriesSDK-6.0.0-0.i386.rpm MQSeriesSamples-6.0.0-0.i386.rpm MQSeriesServer-6.0.0-0.i386.rpm MQSeriesTXClient-6.0.0-0.i386.rpm PreReqs READMEs copyright gsk7bas-7.0-3.15.i386.rpm lap licenses

　　下面是安装步骤：

　　设置Linux服务器。本文使用主机名WMQ1和WMQ2。WMQ1的IP地址为 192.168.28.71，WMQ2的IP地址为192.168.28.72。

　　安装WebSphere MQ前，必须执行./mqlicense.sh，以阅读和接受许可证。

　　安装所需的MQ包及WebSphere MQ Java和IBM Java SDK包，以便为WebSphere MQ开发Java客户机：

以下是引用片段： wmq1:~/mq_install # rpm -ivh IBMJava2-SDK-1.4.2-0.0.i386.rpm wmq1:~/mq_install # rpm -ivh MQSeriesRuntime-6.0.0-0.i386.rpm wmq1:~/mq_install # rpm -ivh MQSeriesServer-6.0.0-0.i386.rpm

　　配置过程包括以下几个步骤：

　　创建队列管理器。

　　启动队列管理器。

　　创建队列和通道。

　　启动命令服务器、侦听器和通道。

　　两台WebSphere MQ计算机都需要执行这些步骤。下面的部分将仅说明如何在WMQ1上进行配置，但在WMQ2上也需要进行这些步骤。

　　创建并启动队列管理器

　　使用root登录到WMQ1，然后键入su - mqm，以更改到用户mqm。使用以下命令为WMQ1计算机创建队列管理器：

以下是引用片段： mqm@wmq1:~/ # crtmqm WMQ1QM

　　使用以下命令启动队列管理器。

以下是引用片段： mqm@wmq1:~/ # strmqm WMQ1QM

　　创建队列和通道

以下是引用片段： 　　使用命令行工具配置WebSphere MQ。以下给出了WMQ1的配置脚本。此脚本定义本地队列、远程队列、传输队列、发送方通道和接收方通道。将以下脚本保存为WMQ1QM.conf： DEFINE QLOCAL ('WMQ1InputQ') + REPLACE DEFINE QLOCAL ('WMQ2QMXmitQ') + DESCR('Transmission Queue to WMQ2QM') + USAGE(XMITQ) + TRIGGER + TRIGTYPE(FIRST) + TRIGDPTH(1) + TRIGDATA('WMQ1QM.WMQ2QM') + INITQ('SYSTEM.CHANNEL.INITQ') + REPLACE DEFINE QREMOTE ('WMQ1OutputQ') + XMITQ('WMQ2QMXmitQ') + RNAME('WMQ2InputQ') + RQMNAME('WMQ2QM') + REPLACE DEFINE CHANNEL ('WMQ1QM.WMQ2QM') CHLTYPE(SDR) + TRPTYPE(TCP) + CONNAME('192.168.28.72(1414)') + XMITQ('WMQ2QMXmitQ') + REPLACE DEFINE CHANNEL ('WMQ2QM.WMQ1QM') CHLTYPE(RCVR) + TRPTYPE(TCP) + REPLACE DEFINE CHANNEL ('SYSTEM.ADMIN.SVRCONN') CHLTYPE(SVRCONN) + TRPTYPE(TCP) + DESCR(' ') + HBINT(300) + MAXMSGL(4194304) + MCAUSER('MUSR_MQADMIN') + RCVDATA(' ') + RCVEXIT(' ') + SCYDATA(' ') + SCYEXIT(' ') + SENDDATA(' ') + SENDEXIT(' ') + REPLACE

　　上面的配置脚本可作为其他WebSphere MQ安装的模版使用。只需要更改粗体部分的名称，即可在其他WebSphere MQ安装中设置队列和通道。此配置脚本中包含大量条目。

　　使用以下命令创建队列和通道，以配置WebSphere MQ：

以下是引用片段： mqm@wmq1:~/ # runmqsc WMQ1QM < WMQ1QM.conf > qcreate.log

　　runmqsc命令用于发出WebSphere MQ命令。在本例中，这些命令从WMQ1QM.conf文件读取，而输出被定向到qcreate.log。打开 qcreate.log，验证其中不存在语法错误，全部有效命令均得到了处理。

　　启动服务

　　需要启动命令服务器、侦听器和通道，然后就可以进行示例应用程序的部署和测试工作了。

以下是引用片段： mqm@wmq1:~/ # strmqcsv WMQ1QM & mqm@wmq1:~/ # runmqlsr -m WMQ1QM -t TCP &

　　验证队列管理器和侦听器在两台计算机上运行，然后启动通道：

以下是引用片段： mqm@wmq1:~/ # runmqchl -m WMQ1QM -c WMQ1QM.WMQ2QM & mqm@wmq1:~/ # runmqchl -m WMQ1QM -c WMQ2QM.WMQ1QM &

　　现在已经安装了WebSphere MQ，并进行了配置，已准备好可以进行测试了。下面一部分将给出两个基于Java的应用程序的源代码，分别用于向队列发送消息和从队列接收消息。

　　示例应用程序

　　用于将消息放入队列中(MQSend)和从队列获取消 (MQGet)的示例Java应用程序可使用前面部分中定义的队列。这些应用程序支持String消息，旨在用于测试和作为说明如何从Java应用程序使用WebSphere MQ的示例。可以从脚本使用应用程序，包括各种类型的应用程序。MQSend和MQGet都从MQConnector类进行扩展，该类可提供WebSphere MQ连接、初始化WebSphere MQ、打开和关闭队列，向队列发送消息及从队列接收消息。

　　MQConnector

　　MQConnector是用于进行发送和获取操作的超类。该类处理打开连接和发送及获取消息的工作。将使用属性文件对其进行配置，需要在属性文件中指定主机地址、队列管理器名称和队列名称：

以下是引用片段： queue.manager=WMQ1QM queue.manager.host=192.168.28.71 queue.name=WMQ1OutputQ 　　以下是MQConnector源代码： package mqconn; import java.io.File; import java.io.FileInputStream; import java.io.IOException; import java.util.Properties; 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; public class MQConnector { protected String qManager = ""; // define name of queue manager protected String qManagerHost = ""; protected String queuName = ""; // define name of queue protected MQQueue mqQueue; protected MQQueueManager qMgr; public static boolean DEBUG = true; public MQConnector() { } public void initMq() { try { FileInputStream fis = new FileInputStream(new File("mqconnect.properties")); Properties props = new Properties(); props.load(fis); fis.close(); qManager = props.getProperty("queue.manager"); qManagerHost = props.getProperty("queue.manager.host"); queuName = props.getProperty("queue.name"); // Create a connection to the queue manager MQEnvironment.channel = "SYSTEM.DEF.SVRCONN"; MQEnvironment.hostname = qManagerHost; debug("Connecting to QueueManager " + qManager + " on " + qManagerHost); qMgr = new MQQueueManager(qManager); } catch (Exception e) { e.printStackTrace(); } } public void openQueue() throws MQException { // Set up the options on the queue we wish to open... // Note. All WebSphere MQ Options are prefixed with MQC in Java. int openOptions = MQC.MQOO_INPUT_AS_Q_DEF | MQC.MQOO_OUTPUT; // Now specify the queue that we wish to open, // and the open options... debug("Opening queue: " + queuName); try { mqQueue = qMgr.accessQueue(queuName, openOptions); } catch(MQException mqe) { //check if MQ reason code 2045 //means that opened queu is remote and it can not be opened as //input queue //try to open as output only if(mqe.reasonCode==2045) { openOptions = MQC.MQOO_OUTPUT; mqQueue = qMgr.accessQueue(queuName, openOptions); } } } public void putMessageToQueue(String msg) throws MQException { try { debug("Sending message: " + msg); MQPutMessageOptions pmo = new MQPutMessageOptions(); MQMessage mqMsg = new MQMessage(); mqMsg.write(msg.getBytes()); // put the message on the queue mqQueue.put(mqMsg, pmo); } catch (IOException e) { e.printStackTrace(); } } public String getMessageFromQueue() throws MQException { try { MQMessage mqMsg = new MQMessage(); MQGetMessageOptions gmo = new MQGetMessageOptions(); // Get a message from the queue mqQueue.get(mqMsg,gmo); //Extract the message data int len=mqMsg.getDataLength(); byte[] message = new byte[len]; mqMsg.readFully(message,0,len); return new String(message); } catch(MQException mqe) { int reason=mqe.reasonCode; if(reason==2033)//no messages { return null; } else { throw mqe; } } catch (IOException e) { e.printStackTrace(); return null; } } public void closeQueue() throws MQException { debug("Closing queue and disconnecting QueueManager..."); // Close the queue... mqQueue.close(); // Disconnect from the queue manager qMgr.disconnect(); } protected boolean hasArg(String arg, String[] args) { for(int i=0;i<args.length;i++) { if(args[i].equals(arg)) { return true; } } return false; } public void debug(Object msg) { if (DEBUG) { System.out.println(msg); } } }

　　该类向其他应用程序提供WebSphere MQ功能。还可以在自己的应用程序中将MQConnector作为其他类的超类或实用类使用。以下方法为Java应用程序提供消息传递功能：

　　initMQ()读取属性并创建队列管理器对象。openQueue()打开属性文件中指定的队列。closeQueue()关闭已打开的队列。disconnectMq()断开队列管理器。putMessageToQueue(String)将字符串消息放置到指定的队列。getMessageFromQueue()从指定的队列读取字符串消息。

　　MQSend

　　MQSend（如下所示）将使用MQConnector类，是用于向队列发送消息的命令行程序。消息作为命令行参数指定，可以使用其将文件内容作为消息发送：

以下是引用片段： package mqconn; import java.io.BufferedReader; import java.io.FileReader; import com.ibm.mq.MQException; public class MQSend extends MQConnector { public MQSend() { } public void send(String[] args) throws MQException { boolean argsAreFiles = hasArg("-f", args); initMq(); openQueue(); for (int i = 0; i < args.length; i++) { if (args[i].equals("-f")) continue; if (!argsAreFiles) { putMessageToQueue(args[i]); } else { try { // send file contents as message BufferedReader br = new BufferedReader(new FileReader(args[i])); StringBuffer msg = new StringBuffer(); for (String line = br.readLine(); line != null; line = br.readLine()) { msg.append(line); msg.append('\n'); } br.close(); putMessageToQueue(msg.toString()); } catch (Exception e) { System.out.println("Error while processing file " + args[i] + ": " + e.toString()); } } System.out.println("Message sent."); } closeQueue(); disconnectMq(); } public static void main(String[] args) { MQSend mqsend = new MQSend(); MQConnector.DEBUG = false; try { if (args == null || args.length == 0) { System.out.println("Usage: " + mqsend.getClass().getName() + " [-f] <file name | message> [<file name | message> ...]"); System.exit(0); } mqsend.send(args); } catch (Exception e) { System.out.println(e.toString()); System.out.println("Usage: " + mqsend.getClass().getName() + " [-f] <file name | message> [<file name | message> ...]"); } } }

　　MQSend使用MQConnector提供消息传递，实际的MQSend程序功能非常简单。它有两种操作模式：

　　将每个命令行参数作为独立消息传递。

　　使用-f选项发送文件内容。在本例中，每个命令行参数都是文件名。

　　MQGet

　　MQGet类（如下所示）非常简单——从指定的队列检索消息，并将其写入到标准输出。可以通过使用重定向将消息存储在文件中。

以下是引用片段： package mqconn; import com.ibm.mq.MQException; public class MQGet extends MQConnector { public MQGet() { } public String getMessages(String[] args) throws MQException { String message=getMessageFromQueue(); return message; } public static void main(String[] args) { MQGet mqget = new MQGet(); MQConnector.DEBUG=false; try { mqget.initMq(); mqget.openQueue(); String msg=mqget.getMessages(args); if(msg!=null) { System.out.println(msg); } mqget.closeQueue(); mqget.disconnectMq(); } catch (Exception e) { e.printStackTrace(); System.out.println("Usage: "+mqget.getClass().getName()+" "); } } }

　　MQGet使用mqconnect.properties文件中指定的队列管理器和队列。下一部分将给出一些可供进行测试的示例用例。

　　示例用例

　　以下是一些用例，均使用WMQ1作为发送方，WMQ2作为接收方，说明了如何使用MQSend和MQGet应用程序。

　　准备工作

　　在使用示例前，您需要进行一些准备工作，如设置属性和编写一些Helper脚本等。为WMQ1中的MQSend应用程序使用以下属性：

以下是引用片段： queue.manager=WMQ1QM queue.manager.host=127.0.0.1 queue.name=WMQ1OutputQ

　　为WMQ2中的MQGet应用程序使用以下属性：

以下是引用片段： queue.manager=WMQ2QM queue.manager.host=127.0.0.1 queue.name=WMQ2InputQ

　　为了节约键入的时间，以下提供了用于进行发送和接收的两个脚本。

以下是引用片段： /opt/IBMJava2-142/bin/java -cp .:com.ibm.mq.jar mqconn.MQSend "$@" /opt/IBMJava2-142/bin/java -cp .:com.ibm.mq.jar mqconn.MQGet "$@"

　　使用chmod 755 mqsend和chmod 755 mqget将这些脚本转换为可执行文件。本文中的所有类和文件都位于/var/mqm/bin目录中，其中/var/mqm是mqm用户的主目录。

　　命令

　　第一个示例命令将消息发送到上面指定的队列。下面的示例清单给出了实际的命令，另外还给出了命令的输出。

以下是引用片段： mqm@wmq1:~/bin> ./mqsend "This is test message" MQJE001: Completion Code 2, Reason 2045 Message sent.

　　MQJE001是预期出现的内容。Reason 2045表示在本例中无法将要打开的队列作为输入队列打开，因为此队列是远程队列。MQJE001由 WebSphere MQ类输出到标准错误。如果不希望看到这些错误消息，请将标准错误重新定向到/dev/null ./mqsend "This is test message" 2> /dev/null。若要了解原因代码的含义，请使用mqrc 2045。

　　在WMQ2上，使用mqget命令接收消息：

以下是引用片段： mqm@wmq2:~/bin> ./mqget This is test message

　　如果再次执行mqget，会看到预期的MQJE001消息。mqrc 2033中的2033表示“MQRC_NO_MSG_AVAILABLE”。检索消息并将其从队列删除。以下代码说明了如何发送多个消息：

以下是引用片段： mqm@wmq1:~/bin> ./mqsend "This is test message" This is another message MQJE001: Completion Code 2, Reason 2045 Message sent. Message sent. Message sent. Message sent. Message sent.

　　接收消息与上面所示的mqget命令相同。以下代码说明了如何接收多个消息：

以下是引用片段： mqm@wmq2:~/bin> ./mqget This is test message mqm@wmq2:~/bin> ./mqget This mqm@wmq2:~/bin> ./mqget is mqm@wmq2:~/bin> ./mqget another mqm@wmq2:~/bin> ./mqget message mqm@wmq2:~/bin> ./mqget MQJE001: Completion Code 2, Reason 2033

　　下一个示例将发送所有Java文件的内容：

以下是引用片段： mqm@wmq1:~/bin> ls -1 mqconn/*.java | awk '{print "mqsend -f "$1""}' | sh 2> /dev/null Message sent. Message sent. Message sent.

　　下一个示例读取消息并将其保存到文件中：

以下是引用片段： mqm@wmq2:~/bin> ./mqget > msg1.txt

　　mqget命令的输出被定向到msg1.txt。

　　结束语

　　本文说明了Linux上的WebSphere MQ使用入门知识。只需要进行一些安装和配置步骤，就能让WebSphere MQ正常运行，而这些步骤正是了解WebSphere MQ的更多信息的良好起点。文中提供的示例应用程序说明了可以如何方便地获取对WebSphere MQ及消息传递功能的访问。示例应用程序还可作为开发自定义应用程序的不错借鉴。

以下是引用片段： wmq1:~/mq_install # rpm -ivh MQSeriesJava-6.0.0-0.i386.rpm

　　WebSphere MQ现在已经完成安装，并有用于进行配置的名为mqm的用户和组。缺省情况下，WebSphere MQ安装在/opt/mqm中，队列和其他配置信息位于用户mqm的主目录中。Java SDK安装在/opt/IBMJava2-142中，Java解释程序位于/opt/IBMJava2-142/bin中。

　　下一步是进行配置。本文为两台计算机使用了具有本地和远程队列的单个队列管理器，用于测试示例应用程序。