WebSphereMQ发送消息和读取消息

我就是我 2022-08-10 03:55 204阅读 0赞

1.webspheremq发送消息

package com.wsmq;
    
    import com.ibm.mq.MQC;
    import com.ibm.mq.MQEnvironment;
    import com.ibm.mq.MQException;
    import com.ibm.mq.MQMessage;
    import com.ibm.mq.MQPutMessageOptions;
    import com.ibm.mq.MQQueue;
    import com.ibm.mq.MQQueueManager;
    
    /**
     * 发送消息程序
     * @author suyunlong
     *
     */
    public class Send 
    {
    	@SuppressWarnings({ "deprecation", "unchecked" })
    	public static void main(String[] args) 
    	{
    		try 
    		{
    			// 主机名称
    			String hostName="192.9.201.150";
    			// 端口（缺省 1414）
    			int port=1414;
    			// 通道名称（缺省）
    			String channel="DC.SVRCONN";
    			// 队列管理器名称
    			String qManager="QM_00000000";
    			// 队列名称 (远程队列名)
    			String qName="RQ_88888888";
    			// 建立MQEnvironment 属性以便客户机连接.
    			MQEnvironment.CCSID=1381;
    			MQEnvironment.hostname=hostName;
    		    MQEnvironment.port=port;
    		    MQEnvironment.channel=channel;
    		    MQEnvironment.properties.put(MQC.TRANSPORT_PROPERTY,MQC.TRANSPORT_MQSERIES); 
    		    // 连接到队列管理器.
    		    MQQueueManager qMgr=new MQQueueManager(qManager);
    		    /**
    		      * 建立打开选项以便打开用于输出的队列，进一步而言，如果队列管理器是
    		      * 停顿的话，我们也已设置了选项去应对不成功情况.
    		      */
    		    int openOptions=MQC.MQOO_OUTPUT | MQC.MQOO_FAIL_IF_QUIESCING;
    		    // 打开队列.
    		    MQQueue queue=qMgr.accessQueue(qName, openOptions, null, null,null);
    		    // 设置放置消息选项我们将使用默认设置.
    		    MQPutMessageOptions pmo=new MQPutMessageOptions();
    		    /**
    		      * 下一步我们建立消息，MQMessage类压缩了包含实际消息数据的数据缓冲区，
    		      * 和描述消息的所有MQMD 参数.
    		      * 欲建立新消息，创建MQMessage类新实例以及使用writxxx(我们将使用writeString 方法.).
    		      * MQQueue 的put()方法也可作为参数MQPutMessageOptions 类的实例.
    		      */
    		    // 创建消息缓冲区.
    		    MQMessage outMsg=new MQMessage();
    		    outMsg.encoding=1381;
    		    outMsg.characterSet=1381;
    		    // 设置MQMD 格式字段.
    		    outMsg.format=MQC.MQFMT_STRING;
    		    // 准备用户数据消息.
    		    String msgString="有志者，事竟成，破釜沉舟，百二秦关终属楚；苦心人，天不负，卧薪尝胆，三千越甲可吞吴。";
    		    // 现在我们在队列上放置消息.
    		    outMsg.writeString(msgString);
    		    // 提交事务处理.
    		    queue.put(outMsg, pmo);
    		    qMgr.commit();
    		    System.out.println(" The message has been Sussesfully put/n/n#########");
    		    // 关闭队列和队列管理器对象.
    		    queue.close();
    		    qMgr.disconnect();
    		} 
    		catch(MQException ex) 
    		{
    			System.out.println("An MQ Error Occurred: Completion Code is :" +
    					ex.completionCode+" The Reason Code is :"+ex.reasonCode);
    			ex.printStackTrace();
    		} 
    		catch(Exception e) 
    		{
    			System.out.println(e.getMessage());
    			e.printStackTrace();
    		}
    	}
    }

2.webspheremq接收消息

package com.wsmq;
    
    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.MQQueue;
    import com.ibm.mq.MQQueueManager;
    
    /**
     * 消息接收器应用程序
     * 调入MQSeries Java API package
     * 为客户机连接设置环境属性
     * 连接到队列管理器
     * 为打开MQSeries 队列设置选项
     * 为获取消息打开应用程序
     * 设置选项，从应用程序队列获取消息
     * 创建消息缓冲区
     * 从队列获取消息到消息缓冲区
     * 从消息缓冲区读取用户数据并在控制台上显示
     */
    public class Receiver 
    {
    	@SuppressWarnings({ "deprecation", "unchecked" })
    	public static void PtpReceiver() 
    	{
    		try 
    		{
    			// 主机名称
    			String hostName="192.9.201.145";
    			// 端口（缺省 1414）
    			int port=1414;
    			// 通道名称（缺省）
    			String channel="DC.SVRCONN";
    			// 队列管理器名称
    			String qManager="QM_00000000";
    			// 队列名称 (本地对列名)
    			String qName="LQ_00000000";
    			// 建立MQEnvironment 属性以便客户机连接
    			MQEnvironment.hostname=hostName;
    			MQEnvironment.port=port;
    			MQEnvironment.channel=channel;
    			MQEnvironment.properties.put(MQC.TRANSPORT_PROPERTY,MQC.TRANSPORT_MQSERIES);
    			MQEnvironment.CCSID=1381; 
    			// 连接到队列管理器.
    			MQQueueManager qMgr=new MQQueueManager(qManager);
    			/**
    		      * 建立打开选项以便打开用于输出的队列，
    		      * 进一步而言，如果队列管理器是停顿的话，我们也
    		      * 已设置了选项去应对不成功情况.
    		      */
    			// 打开队列.
    			int openOptions=MQC.MQOO_INPUT_AS_Q_DEF|MQC.MQOO_INQUIRE|MQC.MQOO_FAIL_IF_QUIESCING;
    			// 设置放置消息选项.
    			MQQueue queue=qMgr.accessQueue(qName, openOptions, null, null, null);
    			if(queue.getCurrentDepth()>0)
    			{
    				MQGetMessageOptions gmo=new MQGetMessageOptions();
    				// 在同步点控制下获取消息.
    				gmo.options=gmo.options + MQC.MQGMO_SYNCPOINT;
    				// 如果在队列上没有消息则等待.
    				gmo.options=gmo.options + MQC.MQGMO_WAIT;
    				// 如果队列管理器停顿则失败.
    				gmo.options=gmo.options + MQC.MQGMO_FAIL_IF_QUIESCING;
    				// 设置等待的时间限制.
    				gmo.waitInterval=3000;
    				/**
    				 * 下一步我们建立消息，MQMessage
    				 * 类压缩了包含实际消息数据的数据缓冲区，
    				 * 和描述消息的所有MQMD 参数.
    				 */
    				// 创建消息缓冲区.
    				MQMessage inMsg=new MQMessage();
    				// 从队列到消息缓冲区获取消息.
    				queue.get(inMsg, gmo);
    				// 从消息读取用户数据.
    				String msgString=inMsg.readStringOfByteLength(inMsg.getMessageLength());
    				// 控制台输出
    				System.out.println("The Message from the Queue is :"+msgString);
    			}
    			// 提交事务处理.
    			qMgr.commit();
    			// 关闭队列和队列管理器对象.
    			queue.close();
    			qMgr.disconnect();
    		} 
    		catch(MQException ex) 
    		{
    			System.out.println("An MQ Error Occurred: Completion Code is :" +
    					ex.completionCode + " The Reason Code is :" + ex.reasonCode);
    			ex.printStackTrace();
    		} 
    		catch(Exception e) 
    		{
    			System.out.println(e.getMessage());
    			e.printStackTrace();
    		}
    	}
    	/**
    	 * @param args
    	 */
    	public static void main(String[] args) 
    	{
    		PtpReceiver();
    	}
    }