扫一扫
分享文章到微信
扫一扫
关注官方公众号
至顶头条
引言
本文的目标是让您轻松实现 Linux® 上的 IBM® WebSphere® MQ 入门。由于大部分 Linux 服务器并没有图形用户界面，因此本文将说明如何使用命令行工具在 Linux 上安装和配置 WebSphere MQ。本文将说明如何使用 WebSphere MQ Java API 创建两个示例 Java™ 应用程序：MQSend（用于向队列发送消息）和 MQGet（用于从队列接收消息）。
先决条件
WebSphere MQ
WebSphere MQ 提供了连接层和应用程序的可靠集成，为 SOA 提供了所需的基础。WebSphere MQ 的主要特色功能包括：
有关 WebSphere MQ 的更多信息，请参见本文结尾的参考资料。
安装 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 |
下面是安装步骤：
WMQ1 和 WMQ2。WMQ1 的 IP 地址为 192.168.28.71，WMQ2 的 IP 地址为 192.168.28.72。
./mqlicense.sh，以阅读和接受许可证。
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 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 中。
下一步是进行配置。本文为两台计算机使用了具有本地和远程队列的单个队列管理器，用于测试示例应用程序。
配置 WebSphere MQ
 |
|
配置过程包括以下几个步骤：
两台 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 信息中心。
使用以下命令创建队列和通道，以配置 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 应用程序提供消息传递功能：
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 程序功能非常简单。它有两种操作模式：
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 及消息传递功能的访问。示例应用程序还可作为开发自定义应用程序的不错借鉴。
如果您非常迫切的想了解IT领域最新产品与技术信息，那么订阅至顶网技术邮件将是您的最佳途径之一。
京公网安备 11010802021500号