WebSphereMq使用JMS发送消息和接收消息

痛定思痛。 2022-08-10 04:50 165阅读 0赞

1.WebSphereMq发送消息

package com.wsmq;
    
    import java.util.Scanner;
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageListener;
    import javax.jms.Queue;
    import javax.jms.QueueConnection;
    import javax.jms.QueueReceiver;
    import javax.jms.QueueSender;
    import javax.jms.QueueSession;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    import com.ibm.mq.jms.MQQueueConnectionFactory;
    
    /**
     * WebSphereMq发送消息
     * @author suyunlong
     *
     */
    public class MQSender implements MessageListener
    {
    	MQQueueConnectionFactory mcf;
    	QueueConnection qconn;
    	final String HOSTNAME="127.0.0.1";
    	final int PORT=1414;
    	final String QUEUEMANAGER_NAME="mq";
    	final String QUEUE_NAME="q1";
    	final String QUEUE_NAME2="q2";
    	boolean replyed=false;
    	private void openConnection() throws JMSException 
    	{
    		mcf=new MQQueueConnectionFactory();
    		mcf.setHostName(HOSTNAME);
    		mcf.setPort(PORT);
    		mcf.setQueueManager(QUEUEMANAGER_NAME);
    		qconn=mcf.createQueueConnection();
    		qconn.start();		
    	}
    	private void sendMessage(String msgInfo) throws JMSException, InterruptedException 
    	{
    		openConnection();
    		QueueSession session=qconn.createQueueSession(false,Session.AUTO_ACKNOWLEDGE);
    		Queue queue=session.createQueue(QUEUE_NAME);
    		Queue responseQueue=session.createQueue(QUEUE_NAME2);
    		QueueSender sender=session.createSender(queue);
    		TextMessage msg=session.createTextMessage(); 
    //        msg.setJMSCorrelationID("123-123456");
    //        msg.setIntProperty("AccountID", 123);
            msg.setJMSReplyTo(responseQueue);   //设置回复队列
            msg.setText(msgInfo);  
    		sender.send(msg); 
    		System.out.println("消息发送:JMSMessage"+msg.getJMSMessageID());
    		//接收回复信息
    		System.out.println("等待客户端回复队列："+msg.getJMSReplyTo());
    		String filter="JMSCorrelationID='"+msg.getJMSMessageID()+"'";  
            QueueReceiver reply=session.createReceiver(responseQueue,filter);
            //同步方式等待接收回复
    //        TextMessage resMsg = (TextMessage) reply.receive(60 * 1000);    
    //        if(resMsg != null){ 
    //			System.out.println("客户端回复消息 : " + resMsg.getText() + " JMSCorrelation" + resMsg.getJMSCorrelationID()); 
    //        }else{
    //        	System.out.println("等待超时！");
    //        }
            //异步方式接收回复
            reply.setMessageListener(this);
            while(!replyed)
            {
            	Thread.sleep(1000);
            }
    		qconn.stop();
    		sender.close();
    		session.close();
    		disConnection();
    	}
    	public void onMessage(Message message) 
    	{
    		try 
    		{
    			String textMessage=((TextMessage)message).getText();
    			System.out.println("客户端回复消息 :"+textMessage+" JMSCorrelation"+message.getJMSCorrelationID()); 			
    		} 
    		catch(JMSException e) 
    		{
    			System.out.println(e.getMessage());
    			e.printStackTrace();
    		}
    		finally
    		{
    			replyed=true;	
    		}
    	}
    	private void disConnection() throws JMSException 
    	{     	
    		qconn.close(); 
    	}
    	@SuppressWarnings("resource")
    	public static void main(String[] args) throws JMSException, InterruptedException 
    	{
    		MQSender ms=new MQSender();
    		Scanner scan=new Scanner(System.in);
    		System.out.print("输入信息：");
    		ms.sendMessage(scan.next());
    		System.out.println("消息发送完毕！");
    	}
    }

2.WebSphereMq接收消息

package com.wsmq;
    
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageListener;
    import javax.jms.Queue;
    import javax.jms.QueueConnection;
    import javax.jms.QueueReceiver;
    import javax.jms.QueueSession;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    import com.ibm.mq.jms.MQQueueConnectionFactory;
    
    /**
     * WebSphereMq接收消息
     * @author suyunlong
     *
     */
    public class MQRecevicer implements MessageListener
    {
    	MQQueueConnectionFactory mcf;
    	QueueConnection qconn;
    	QueueSession session;
    	final String HOSTNAME="127.0.0.1";
    	final int PORT=1414;
    	final String QUEUEMANAGER_NAME="mq";
    	final String QUEUE_NAME="q1";
    	final String QUEUE_NAME2="q2";
    	boolean replyed=false;
    	public void openConnection() throws JMSException 
    	{
    		mcf=new MQQueueConnectionFactory();
    		mcf.setHostName(HOSTNAME);
    		mcf.setPort(PORT);
    		mcf.setQueueManager(QUEUEMANAGER_NAME);
    		qconn=mcf.createQueueConnection();
    		qconn.start();		
    	}
    	public void disConnection() throws JMSException 
    	{     	
            qconn.close(); 
    	}
    	public void recevicerMessage(String reply) throws JMSException, InterruptedException 
    	{
    		openConnection(); 
    		session=qconn.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
    		Queue queue=session.createQueue(QUEUE_NAME);
    		QueueReceiver recevier=session.createReceiver(queue);
    		//同步方式接收消息并回复
    //		TextMessage textMessage = (TextMessage) recevier.receive();
    //		System.out.println("接收消息 : " + textMessage.getText() + "  JMSMessage" +textMessage.getJMSMessageID()); 
    //		Queue responseQueue = (Queue) textMessage.getJMSReplyTo();  
    //		if(responseQueue != null){
    //			TextMessage responseMsg = session.createTextMessage();
    //			responseMsg.setJMSCorrelationID(textMessage.getJMSMessageID());
    //			responseMsg.setText("This message is reply from client..."+textMessage.getText());
    //	        session.createSender(responseQueue).send(responseMsg); 
    //	        System.out.println("客户端回复队列："+responseQueue.toString()+"  JMSCorrelation"+responseMsg.getJMSCorrelationID());  
    //		}else{
    //			System.out.println("服务端回复队列为空");  
    //		}
    		//异步方式接收消息并回复
    		recevier.setMessageListener(this);
    		while(!replyed)
    		{
    			Thread.sleep(1000);
    		}
    		qconn.stop();
    		recevier.close();
    		session.close();
    		disConnection();
    	}
    	public void onMessage(Message message) 
    	{
    		try 
    		{
    			String textMessage=((TextMessage)message).getText();
    			System.out.println("接收消息 : "+textMessage+"  JMSMessage"+message.getJMSMessageID()); 
    			Queue responseQueue=(Queue)message.getJMSReplyTo();  
    			if(responseQueue != null)
    			{
    				TextMessage responseMsg=session.createTextMessage();
    				responseMsg.setJMSCorrelationID(message.getJMSMessageID());
    				responseMsg.setText("This message is reply from client："+textMessage);
    		        session.createSender(responseQueue).send(responseMsg); 
    		        System.out.println("客户端回复队列："+responseQueue+"  JMSCorrelation"+responseMsg.getJMSCorrelationID());  
    			}
    			else
    			{
    				System.out.println("服务端回复队列为空");  
    			}
    		} 
    		catch(JMSException e) 
    		{
    			System.out.println(e.getMessage());
    			e.printStackTrace();
    		}
    		finally
    		{
    			replyed=true;
    		}
    	}
    	public static void main(String[] args) throws JMSException, InterruptedException 
    	{
    		MQRecevicer mr=new MQRecevicer();
    		System.out.println("正在接收消息...");
    		mr.recevicerMessage("消息已经收到，这是接收端的回复！");
    		System.out.println("消息接收完毕！");
    	}
    }