JAVA如何获取mq 发送的信息

获取信息时报错：MQJE001: 完成代码为“2”，原因为“2045”。A WebSphere MQ error occurred : Completion code 2 Reason code 2045

下面是我的测试环境

队列管理器和队列配置

通道配置

代码：

/**

* 从队列中去获取消息，如果队列中没有消息，就会发生异常，不过没有关系，有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;

}

使用 JAVA 语言自己动手来写一个MQ (类似ActiveMQ,RabbitMQ)

主要角色

首先我们必须需要搞明白 MQ (消息队列) 中的三个基本角色

Producer

Broker

Consumer

整体架构如下所示

自定义协议

首先从上一篇中介绍了协议的相关信息,具体厂商的 MQ(消息队列) 需要遵循某种协议或者自定义协议 , 消息的 生产者和消费者需要遵循其协议(约定)才能后成功地生产消息和生产消息 ,所以在这里我们自定义一个协议如下．

消息处理中心 : 如果接收到的信息包含"SEND"字符串,即视为生产者发送的消息,消息处理中心需要将此信息存储等待消费者消费

消息处理中心 : 如果接受到的信息为CONSUME，既视为消费者发送消费请求，需要将存储的消息队列头部的信息转发给消费者，然后将此消息从队列中移除

消息处理中心 : 如果消息处理中心存储的消息满3条仍然没有消费者进行消费,则不再接受生产者的生产请求

消息生产者：需要遵循协议将生产的消息头部增加＂SEND:＂ 表示生产消息

消息消费者：需要遵循协议向消息处理中心发送＂CONSUME＂字符串表示消费消息

流程顺序

项目构建流程

下面将整个ＭＱ的构建流程过一遍

新建一个 Broker 类，内部维护一个 ArrayBlockingQueue 队列，提供生产消息和消费消息的方法， 仅仅具备存储服务功能

新建一个 BrokerServer 类,将 Broker 发布为服务到本地9999端口，监听本地9999端口的 Socket 链接，在接受的信息中进行我们的协议校验, 这里 仅仅具备接受消息,校验协议,转发消息功能;

新建一个 MqClient 类,此类提供与本地端口9999的Socket链接 , 仅仅具备生产消息和消费消息的方法

测试：新建两个 MyClient 类对象，分别执行其生产方法和消费方法

具体使用流程

生产消息：客户端执行生产消息方法，传入需要生产的信息，该信息需要遵循我们自定义的协议，消息处理中心服务在接受到消息会根据自定义的协议校验该消息是否合法，如果合法如果合法就会将该消息存储到Broker内部维护的 ArrayBlockingQueue 队列中．如果 ArrayBlockingQueue 队列没有达到我们协议中的最大长度将将消息添加到队列中，否则输出生产消息失败．

消息消息：客户端执行消费消息方法， Broker服务 会校验请求的信息的信息是否等于 CONSUME ，如果验证成功则从Broker内部维护的 ArrayBlockingQueue 队列的 Poll 出一个消息返回给客户端

代码演示

消息处理中心 Broker

/**

* 消息处理中心

*/

public class Broker {

// 队列存储消息的最大数量

private final static int MAX_SIZE = 3;

// 保存消息数据的容器

private static ArrayBlockingQueue messageQueue = new ArrayBlockingQueue(MAX_SIZE);

// 生产消息

public static void produce(String msg) {

if (messageQueue.offer(msg)) {

System.out.println("成功向消息处理中心投递消息：" + msg + "，当前暂存的消息数量是：" + messageQueue.size());

} else {

System.out.println("消息处理中心内暂存的消息达到最大负荷，不能继续放入消息！");

}

System.out.println("=======================");

}

// 消费消息

public static String consume() {

String msg = messageQueue.poll();

if (msg != null) {

// 消费条件满足情况，从消息容器中取出一条消息

System.out.println("已经消费消息：" + msg + "，当前暂存的消息数量是：" + messageQueue.size());

} else {

System.out.println("消息处理中心内没有消息可供消费！");

}

System.out.println("=======================");

return msg;

}

}

消息处理中心服务 BrokerServer

/**

* 用于启动消息处理中心

*/

public class BrokerServer implements Runnable {

public static int SERVICE_PORT = 9999;

private final Socket socket;

public BrokerServer(Socket socket) {

this.socket = socket;

}

@Override

public void run() {

try (

BufferedReader in = new BufferedReader(new InputStreamReader(

socket.getInputStream()));

PrintWriter out = new PrintWriter(socket.getOutputStream())

)

{

while (true) {

String str = in.readLine();

if (str == null) {

continue;

}

System.out.println("接收到原始数据：" + str);

if (str.equals("CONSUME")) { //CONSUME 表示要消费一条消息

//从消息队列中消费一条消息

String message = Broker.consume();

out.println(message);

out.flush();

} else if (str.contains("SEND:")){

//接受到的请求包含SEND:字符串 表示生产消息放到消息队列中

Broker.produce(str);

}else {

System.out.println("原始数据:"+str+"没有遵循协议,不提供相关服务");

}

}

} catch (Exception e) {

e.printStackTrace();

}

}

public static void main(String[] args) throws Exception {

ServerSocket server = new ServerSocket(SERVICE_PORT);

while (true) {

BrokerServer brokerServer = new BrokerServer(server.accept());

new Thread(brokerServer).start();

}

}

}

客户端 MqClient

/**

* 访问消息队列的客户端

*/

public class MqClient {

//生产消息

public static void produce(String message) throws Exception {

//本地的的BrokerServer.SERVICE_PORT 创建SOCKET

Socket socket = new Socket(InetAddress.getLocalHost(), BrokerServer.SERVICE_PORT);

try (

PrintWriter out = new PrintWriter(socket.getOutputStream())

) {

out.println(message);

out.flush();

}

}

//消费消息

public static String consume() throws Exception {

Socket socket = new Socket(InetAddress.getLocalHost(), BrokerServer.SERVICE_PORT);

try (

BufferedReader in = new BufferedReader(new InputStreamReader(

socket.getInputStream()));

PrintWriter out = new PrintWriter(socket.getOutputStream())

) {

//先向消息队列发送命令

out.println("CONSUME");

out.flush();

//再从消息队列获取一条消息

String message = in.readLine();

return message;

}

}

}

测试MQ

public class ProduceClient {

public static void main(String[] args) throws Exception {

MqClient client = new MqClient();

client.produce("SEND:Hello World");

}

}

public class ConsumeClient {

public static void main(String[] args) throws Exception {

MqClient client = new MqClient();

String message = client.consume();

System.out.println("获取的消息为：" + message);

}

}

我们多执行几次客户端的生产方法和消费方法就可以看到一个完整的MQ的通讯过程,下面是我执行了几次的一些日志

接收到原始数据：SEND:Hello World

成功向消息处理中心投递消息：SEND:Hello World，当前暂存的消息数量是：1

=======================

接收到原始数据：SEND:Hello World

成功向消息处理中心投递消息：SEND:Hello World，当前暂存的消息数量是：2

=======================

接收到原始数据：SEND:Hello World

成功向消息处理中心投递消息：SEND:Hello World，当前暂存的消息数量是：3

=======================

接收到原始数据：SEND:Hello World

消息处理中心内暂存的消息达到最大负荷，不能继续放入消息！

=======================

接收到原始数据：Hello World

原始数据:Hello World没有遵循协议,不提供相关服务

接收到原始数据：CONSUME

已经消费消息：SEND:Hello World，当前暂存的消息数量是：2

=======================

接收到原始数据：CONSUME

已经消费消息：SEND:Hello World，当前暂存的消息数量是：1

=======================

接收到原始数据：CONSUME

已经消费消息：SEND:Hello World，当前暂存的消息数量是：0

=======================

接收到原始数据：CONSUME

消息处理中心内没有消息可供消费！

=======================

小结

本章示例代码主要源自分布式消息中间件实践一书 , 这里我们自己使用Java语言写了一个MQ消息队列 , 通过这个消息队列我们对MQ中的几个角色 "生产者,消费者,消费处理中心,协议" 有了更深的理解 ; 那么下一章节我们就来一块学习具体厂商的MQ RabbitMQ

IBM MQ 6.0中设置两个队列，(远程队列、通道之类都不设置)。

队列管理器是XIR_QM_1502

队列名称是ESBREQ

IP地址是10.23.117.134(远程的一台电脑，跟我的电脑不在一个局域网内)

端口1414

CCSID 1208

MQ配置可以参考这个，有配图http://wenku.baidu.com/view/06d108d0360cba1aa811daa3.html

程序如下，发送线程两个，接收线程一个。接收完毕后就结束。

/*

* 创建日期 2012-7-10

*

* TODO 要更改此生成的文件的模板，请转至

* 窗口 － 首选项 － Java － 代码样式 － 代码模板

*/

package yerasel;

/**

* @author Fenglb E-mail:56553655@163.com

* @version 创建时间：2009-4-30 下午04:13:38 类说明

*/

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.MQPutMessageOptions;

import com.ibm.mq.MQQueue;

import com.ibm.mq.MQQueueManager;

interface SomeConstants {

String qManager = "XIR_QM_1502";//"XIR_QM"; //QueueManager name

String qName = "ESBREQ";// Queue Name

String strIP = "10.23.117.134";//"10.24.28.139";//"10.24.28.102";

int iPort = 1502;//1414;

String strChl = "SYSTEM.DEF.SVRCONN";// Server-Connection Channel

int iCCSID = 1208;

}

class Sender implements Runnable, SomeConstants {

public void run() {

sendMessage();

}

public void sendMessage() {

String name = Thread.currentThread().getName();

System.out.println("进入线程" + name);

MQQueueManager qMgr = null;

// configure connection parameters

MQEnvironment.hostname = strIP;

// Server name or IP

MQEnvironment.port = iPort;

MQEnvironment.channel = strChl;

MQEnvironment.CCSID = iCCSID;

// java程序连接mq的方式有两种，一是客户机方式，一是绑定方式，

// 默认是客户机方式，当mq部署在本地的时候，就需要用绑定方式

// 本机IP是10.24.28.139连接10.23.117.134的时候不需要下句

//MQEnvironment.properties.put(MQC.TRANSPORT_PROPERTY,

//MQC.TRANSPORT_MQSERIES_BINDINGS);

// Create a connection to the QueueManager

System.out.println(name + " Connecting to queue manager: " + qManager);

try {

qMgr = new MQQueueManager(qManager);

// Set up the options on the queue we wish to open

int openOptions = MQC.MQMT_REQUEST | MQC.MQPMO_NEW_MSG_ID

| MQC.MQOO_OUTPUT | MQC.MQOO_FAIL_IF_QUIESCING

| MQC.MQOO_INPUT_AS_Q_DEF;

// Now specify the queue that we wish to open and the open options

System.out.println(name + " Accessing queue: " + qName);

MQQueue queue = qMgr.accessQueue(qName, openOptions);

// Define a simple WebSphere MQ Message ...

// Specify the default put message options

MQPutMessageOptions pmo = new MQPutMessageOptions();

// Put the message to the queue

System.out.println(name + " Sending a message...");

MQMessage msg = new MQMessage();

msg.messageId = "MSGID".getBytes();

msg.messageType = MQC.MQMT_REQUEST;

msg.replyToQueueName = "ESBREQ";

// 在此测试一下 mq 的传输次列

for (int j = 1; j < 5; j++) {

msg.messageSequenceNumber = j;

// write some text in UTF8 format

try {

String str = "Salemetsizbe Yerasel";

str = str + " " + j;

msg.writeUTF(str);

queue.put(msg, pmo);

msg.clearMessage();

System.out.println(name + " putting the message... " + j);

} catch (MQException mqe) {

mqe.printStackTrace();

break;

} catch (IOException e1) {

e1.printStackTrace();

}

}

qMgr.commit();

System.out.println(name + " Done!");

System.out.println("==========");

System.out.println("");

} catch (MQException e) {

e.printStackTrace();

}

}

}

class Receiver implements Runnable, SomeConstants {

public void run() {

recvMessage();

}

public void recvMessage() {

String name = Thread.currentThread().getName();

try {

Thread.sleep(1000);

MQQueueManager qMgr = null;

System.out.println("进入线程" + name);

System.out.println(name + " Connecting to queue manager: "

+ qManager);

qMgr = new MQQueueManager(qManager);

// 设置将要连接的队列属性

// 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

| MQC.MQOO_INQUIRE;

// Now get the message back again. First define a WebSphere MQ

// message to receive the data

MQMessage rcvMessage = new MQMessage();

// Specify default get 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.(设置等待的毫秒时间限制)

System.out.println(name + " Accessing queue: " + qName);

MQQueue queue = qMgr.accessQueue(qName, openOptions);

int depth = 0;

// Get the message off the queue.

System.out.println("... " + name + " getting the message back again");

for (;;) {

try {

queue.get(rcvMessage, gmo);

System.out.println(" ID: "

+ (new String(rcvMessage.messageId)).trim()

+ " Num: " + rcvMessage.messageSequenceNumber

+ " Type: " + rcvMessage.messageType + " Flag: "

+ rcvMessage.messageFlags);

// And display the message text...

String msgText = rcvMessage.readUTF();

System.out.println("The message is: " + msgText);

rcvMessage.clearMessage();

// Break if no MSG left in queue

depth = queue.getCurrentDepth();

if (depth == 0)

break;

} catch (MQException mqe) {

mqe.printStackTrace();

break;

// null;

} catch (IOException e) {

e.printStackTrace();

}

}

// Close the queue

System.out.println(name + " Closing the queue");

queue.close();

// Disconnect from the QueueManager

System.out.println(name + " Disconnecting from the Queue Manager");

qMgr.disconnect();

System.out.println(name + " Done!");

System.out.println("==========");

System.out.println("");

} catch (MQException ex) {

System.out

.println("A WebSphere MQ Error occured : Completion Code "

+ ex.completionCode + " Reason Code "

+ ex.reasonCode + ex.getMessage());

} catch (InterruptedException e1) {

e1.printStackTrace();

}

}

}

public class MQTest {

public static void main(String args[]) {

/*

* MQTest first = new MQTest(); first.sendMessage();

* first.recvMessage();

*/

Sender sender = new Sender();

Thread senderThread = new Thread(sender);

senderThread.start();

senderThread.setName("Sender");

Thread senderThread2 = new Thread(sender);

senderThread2.start();

senderThread2.setName("Sender2");

Receiver recv = new Receiver();

Thread recvThread = new Thread(recv);

recvThread.start();

recvThread.setName("Receiver");

// Receiver recv = new Receiver();

// new Thread(recv).start();

}

}

运行结果如下：

进入线程Sender2

进入线程Sender

Sender2 Connecting to queue manager: XIR_QM_1502

Sender Connecting to queue manager: XIR_QM_1502

Sender2 Accessing queue: ESBREQ

Sender2 Sending a message...

Sender Accessing queue: ESBREQ

Sender Sending a message...

Sender2 putting the message... 1

Sender putting the message... 1

Sender2 putting the message... 2

Sender putting the message... 2

Sender2 putting the message... 3

Sender putting the message... 3

Sender2 putting the message... 4

Sender putting the message... 4

Sender2 Done!

==========

Sender Done!

==========

进入线程Receiver

Receiver Connecting to queue manager: XIR_QM_1502

Receiver Accessing queue: ESBREQ

... Receiver getting the message back again

ID: MSGID Num: 1 Type: 1 Flag: 0

The message is: Salemetsizbe Yerasel 1

ID: MSGID Num: 1 Type: 1 Flag: 0

The message is: Salemetsizbe Yerasel 1

ID: MSGID Num: 1 Type: 1 Flag: 0

The message is: Salemetsizbe Yerasel 2

ID: MSGID Num: 1 Type: 1 Flag: 0

The message is: Salemetsizbe Yerasel 2

ID: MSGID Num: 1 Type: 1 Flag: 0

The message is: Salemetsizbe Yerasel 3

ID: MSGID Num: 1 Type: 1 Flag: 0

The message is: Salemetsizbe Yerasel 3

ID: MSGID Num: 1 Type: 1 Flag: 0

The message is: Salemetsizbe Yerasel 4

ID: MSGID Num: 1 Type: 1 Flag: 0

The message is: Salemetsizbe Yerasel 4

Receiver Closing the queue

Receiver Disconnecting from the Queue Manager

Receiver Done!

==========