IBM WebSphere MQ发送消息和接收消息

Dear 丶 2022-08-18 13:30 275阅读 0赞

package com.imp;
    
    import java.io.FileOutputStream;
    import java.util.ArrayList;
    import java.util.List;
    import com.ibm.mq.MQC;
    import com.ibm.mq.MQEnvironment;
    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;
    import com.util.BCDUtil;
    import com.util.LoggerInfo;
    
    /**
     * IBM WebSphere MQ工厂类
     * @author suyunlong
     *
     */
    public class MQFactory 
    {
    	private MQQueueManager mgr;
    	/**
    	 * 打开到MQ服务器的连接
    	 * Client模式
    	 * @param MQ_HOST
    	 * @param MQ_PORT
    	 * @param MQ_CHANNEL
    	 * @param MQ_QUEUE_MANAGER
    	 */
    	public void openConnection(String MQ_HOST,int MQ_PORT,String MQ_CHANNEL,
    			String MQ_QUEUE_MANAGER)
    	{
    		MQEnvironment.hostname=MQ_HOST;
    		MQEnvironment.channel=MQ_CHANNEL;
    		//MQEnvironment.CCSID=1381; // 字符编码
    		MQEnvironment.port=MQ_PORT;
    		try 
    		{
    			mgr=new MQQueueManager(MQ_QUEUE_MANAGER);
    		}
    		catch(Exception e) 
    		{
    			LoggerInfo.logerror.error(e.getMessage());
    			System.out.println(e.getMessage());
    			e.printStackTrace();
    		}
    	}
    	/**
    	 * 打开到MQ服务器的连接
    	 * Binding模式
    	 * @param MQ_QUEUE_MANAGER
    	 */
    	public void openConnection(String MQ_QUEUE_MANAGER)
    	{
    		try 
    		{
    			mgr=new MQQueueManager(MQ_QUEUE_MANAGER);
    		}
    		catch(Exception e) 
    		{
    			LoggerInfo.logerror.error(e.getMessage());
    			System.out.println(e.getMessage());
    			e.printStackTrace();
    		}
    	}
    	/**
    	 * 发送消息到MQ
    	 * @param MQ_QUEUE
    	 * @param message
    	 */
    	public void sendMessage(String MQ_QUEUE,String message) 
    	{
    		MQQueue mqQueue=null;
    		try 
    		{
    			int openOptions=MQC.MQOO_INPUT_AS_Q_DEF|MQC.MQOO_OUTPUT|MQC.MQOO_INQUIRE;
    			mqQueue=mgr.accessQueue(MQ_QUEUE,openOptions,null,null,null);
    			MQPutMessageOptions pmo=new MQPutMessageOptions();
    			MQMessage msg=new MQMessage();
    			String mid="0LWZZ";
    			msg.messageId=mid.getBytes();
    			msg.messageType=MQC.MQMT_DATAGRAM;
    			msg.persistence=MQC.MQPER_PERSISTENCE_AS_Q_DEF;
    			//msg.format=MQC.MQFMT_STRING;
    			//msg.characterSet=1381;
    			// 设置消息用不过期
    			msg.expiry=-1; 
    			List<Byte> list=new ArrayList<Byte>();
    			list.add((byte)0x01);
    			list.add((byte)0x00);
    			list.add((byte)0x00);
    			list.add((byte)0x10);
    			List<Byte> mlist=new ArrayList<Byte>();
    			mlist.add((byte)0xAB);
    			mlist.add((byte)0x7B);
    			mlist.add((byte)0x00);
    			mlist.add((byte)0x10);
    			mlist.add((byte)0x00);
    			mlist.add((byte)0x00);
    			mlist.add((byte)0x03);
    			mlist.add((byte)0xE8);
    			List<Byte> data=new ArrayList<Byte>();
    			String str="28569283,"+"28569271,"+"6071,"+"22,"+"121164,"+"605,"+"0.00,"+"1200,"+
    						"102,"+"1,"+"2015-11-10 下午 04:23:15,"+"C020,"+"乌海北,"+"102,"+
    						"27	A5 5A 16 22 00 00 60 "+"71 12 11 64 06 05 12 00 15 11 10 16 24 53 " +
    						"00 66 35 FA AD ED,"+"1447144875";
    			byte bt[]=str.getBytes();
    			for(int i=0;i<bt.length;i++)
    			{
    				data.add(bt[i]);
    			}
    			int length=data.size();
    			byte len[]=BCDUtil.getByteGWDW(length);
    			mlist.add(len[0]);
    			mlist.add(len[1]);
    			mlist.addAll(data);
    			mlist.add((byte)0xAB);
    			mlist.add((byte)0x7B);
    			byte mlen[]=BCDUtil.getByteGWDW(mlist.size());
    			list.add(mlen[0]);
    			list.add(mlen[1]);
    			list.addAll(mlist);
    			List<Byte> pack=new ArrayList<Byte>();
    			String version="V1";
    			byte b[]=version.getBytes();
    			for(int i=0;i<b.length;i++)
    			{
    				pack.add(b[i]);
    			}
    			pack.add((byte)0x20);
    			int l=1000;
    			int bs=0;
    			if(list.size()%l==0)
    			{
    				bs=list.size()/l;
    			}
    			else
    			{
    				bs=list.size()/l+1;
    			}
    			String s=bs+"";
    			b=s.getBytes();
    			for(int i=0;i<b.length;i++)
    			{
    				pack.add(b[i]);
    			}
    			pack.add((byte)0x20);
    			int size=list.size();
    			for(int i=0;i<bs;i++)
    			{
    				int tmp=0;
    				if(size-l>=0)
    				{
    					tmp=l;
    				}
    				else
    				{
    					tmp=size;
    				}
    				size=size-tmp;
    				s=tmp+"";
    				b=s.getBytes();
    				for(int j=0;j<b.length;j++)
    				{
    					pack.add(b[j]);
    				}
    				if(i==bs-1)
    				{
    					pack.add((byte)0x0A);
    				}
    				else
    				{
    					pack.add((byte)0x20);
    				}
    			}
    			int fromIndex=0;
    			int toIndex=0;
    			size=list.size();
    			for(int i=0;i<bs;i++)
    			{
    				int tmp=0;
    				if(size-l>=0)
    				{
    					tmp=l;
    				}
    				else
    				{
    					tmp=size;
    				}
    				size=size-tmp;
    				toIndex=toIndex+tmp;
    				List<Byte> t=list.subList(fromIndex,toIndex);
    				pack.addAll(t);
    				fromIndex=fromIndex+tmp;
    			}
    			byte bb[]=new byte[pack.size()];
    			for(int i=0;i<pack.size();i++)
    			{
    				bb[i]=pack.get(i);
    			}
    			msg.write(bb);
    			// 将消息放入队列
    			mqQueue.put(msg,pmo);
    			LoggerInfo.logerror.info("发送消息成功！");
    			FileOutputStream fos=new FileOutputStream("E:\\share\\12.txt");
    			fos.write(bb);  
    			fos.close();
    		} 
    		catch(Exception e) 
    		{
    			LoggerInfo.logerror.error(e.getMessage());
    			System.out.println(e.getMessage());
    			e.printStackTrace();
    		} 
    		finally
    		{
    			try 
    			{
    				if(mqQueue!=null)
    				{
    					mqQueue.close();
    					mqQueue=null;
    				}
    			}
    			catch(Exception e) 
    			{
    				LoggerInfo.logerror.error(e.getMessage());
    				System.out.println(e.getMessage());
    				e.printStackTrace();
    			}
    		}
    	}
    	/**
    	 * 读取MQ消息
    	 * @param MQ_QUEUE
    	 */
    	public void receiveMessage(String MQ_QUEUE)
    	{
    		MQQueue mqQueue=null;
    		try 
    		{
    			int openOptions=MQC.MQOO_INPUT_AS_Q_DEF|MQC.MQOO_OUTPUT|MQC.MQOO_INQUIRE;
    			mqQueue=mgr.accessQueue(MQ_QUEUE,openOptions,null,null,null);
    			MQGetMessageOptions gmo=new MQGetMessageOptions();
    			MQMessage msg=new MQMessage();
    			if(mqQueue.getCurrentDepth()>0)
    			{
    				mqQueue.get(msg,gmo);
    				int length=msg.getDataLength();
    				byte b[]=new byte[length];
    				System.out.println("Message Length:"+msg.getDataLength()+" "+
    						msg.getMessageLength());
    				msg.readFully(b,0,length);
    				//String filename=TimeUtil.getCurrentDate();
    				FileOutputStream fos=new FileOutputStream("E:\\share\\12.txt");
    				fos.write(b);
    				fos.close();
    			}
    		} 
    		catch(Exception e) 
    		{
    			LoggerInfo.logerror.error(e.getMessage());
    			System.out.println(e.getMessage());
    			e.printStackTrace();
    		} 
    		finally
    		{
    			try 
    			{
    				if(mqQueue!=null)
    				{
    					mqQueue.close();
    					mqQueue=null;
    				}
    			}
    			catch(Exception e) 
    			{
    				LoggerInfo.logerror.error(e.getMessage());
    				System.out.println(e.getMessage());
    				e.printStackTrace();
    			}
    		}
    	}
    	/**
    	 * 关闭连接
    	 */
    	public void closeConnection() 
    	{
    		try 
    		{
    			if(mgr!=null) 
    			{
    				mgr.close();
    				mgr=null;
    			}
    		}
    		catch(Exception e) 
    		{
    			LoggerInfo.logerror.error(e.getMessage());
    			System.out.println(e.getMessage());
    			e.printStackTrace();
    		}
    	}
    	public static void main(String args[])
    	{
    		String MQ_HOST="10.1.227.114";
    		int MQ_PORT=1414;
    		String MQ_CHANNEL="LWZZ.CHL";
    		String MQ_QUEUE_MANAGER="LWZZAM";
    		MQFactory mq=new MQFactory();
    		mq.openConnection(MQ_HOST, MQ_PORT, MQ_CHANNEL, MQ_QUEUE_MANAGER);
    		mq.receiveMessage("LWZZFROMGPRSQ");
    		//mq.sendMessage("LWZZTOGPRSQ", "");
    		mq.closeConnection();
    	}
    }