Apache ActiveMQ P2P模式实时请求-响应

目标：使用开源JMS应用框架Apache ActiveMQ，实现P2P方式下Request-Response Socket通信

模式：P2P

特征：发送端可以实时发送请求并取得响应，接收端可以实时获取请求并返回响应

库：activemq-all-5.11.1.jar

前提：启动activemq消息服务程序

--------------------------------------------------------------------------------------------

发送端：

package merrick.activemq;

import java.util.UUID;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

public class Sender_p2p {
 
 private static String user = ActiveMQConnection.DEFAULT_USER; 
    private static String password =ActiveMQConnection.DEFAULT_PASSWORD; 
    private static String url =  "tcp://localhost:61616";  
   
    public static final String LONGMSG = "123456789012345678901234567890123456789012345678901234567890";

 public static void main(String[] args) {  

  try {
   
   new Sender_p2p().send(UUID.randomUUID().toString()+LONGMSG);
   
   
  } catch (Throwable e) {
   
   e.printStackTrace();
  }
  

 }
 
 /***
  * P2P模式
  * Request-Response
  * 类似：发送请求同步获取响应
  * 此处可应用为TCP SOCKET Client端
  *
  * */ 
 public  void send(String msg) throws Exception {

      ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user,password,url);
      Connection connection = connectionFactory.createConnection(); 
      connection.start();       
      Session session = connection.createSession(Boolean.TRUE,Session.AUTO_ACKNOWLEDGE); 
      Queue  destination = session.createQueue("Request&Response_X"); 
      MessageProducer producer = session.createProducer(destination);       
      producer.setDeliveryMode(DeliveryMode.PERSISTENT);  
      TextMessage message = session.createTextMessage(msg);
 
     
      Destination tempDest = session.createTemporaryQueue(); 
      MessageConsumer responseConsumer = session.createConsumer(tempDest); 
      message.setJMSReplyTo(tempDest); 
      String correlationId = UUID.randomUUID().toString();      
      message.setJMSCorrelationID(correlationId);  //
      responseConsumer.setMessageListener(new MessageListener() {//监听回应
  @Override
  public void onMessage(Message arg0) {   
   try {
    TextMessage textMessage=(TextMessage)arg0; 
    System.out.println(textMessage.getText()); //显示接收者的Response
   } catch (JMSException e) {    
    e.printStackTrace();
   } finally {//得到回应后关闭会话和连接
                try {
      session.commit();
      session.close();
      connection.close();
     } catch (JMSException e) {

      e.printStackTrace();
     }
   }
   
  }});
     
      producer.send(message); //发送消息     
    
      System.out.println("<"+Thread.currentThread().getId() + "> send ok: "+msg);
      session.commit(); 
 }

}

 

--------------------------------------------------------------------------------------------

接收端：

package merrick.activemq;

import java.util.UUID;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

public class Receiver_p2p {
 
 
 private static String user = ActiveMQConnection.DEFAULT_USER; 
    private static String password =ActiveMQConnection.DEFAULT_PASSWORD; 
    private static String url = "tcp://localhost:61616"; 
   
   

 public static void main(String[] args) {
  
  try {
   Receiver_p2p obj = new Receiver_p2p();
   obj.receive();
   
  } catch (Exception e) {

   e.printStackTrace();
  }
 }
 
 /***
  * P2P模式
  * Request-Response
  * 类似：获取即时内容，同步返回响应
  * (但Session不能关闭)
  * 此处可应用为TCP SOCKET Server端
  * */ 
 public void receive() throws Exception{
  ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user,password,url); 
  Connection connection = connectionFactory.createConnection(); 
  connection.start(); 
  Session session = connection.createSession(Boolean.TRUE,Session.AUTO_ACKNOWLEDGE);
  
  Queue  destination = session.createQueue("Request&Response_X");
  MessageConsumer consumer = session.createConsumer(destination);
  consumer.setMessageListener(new MessageListener() {
   
   @Override
   public void onMessage(Message message) {  //监听消息  
    
                try {
                 TextMessage textMessage=(TextMessage)message; 
     System.out.println("[Received]"+textMessage.getText()); //取得消息
     
     
     Message response = session.createTextMessage(UUID.randomUUID().toString() + " [from receiver] "+ textMessage.getText().length()+", " + textMessage.getText());//及时给予Response
     response.setJMSCorrelationID(message.getJMSCorrelationID());     
     MessageProducer producer = session.createProducer(null); 
     producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
     producer.send(message.getJMSReplyTo(),response);  //发送回应             
     
    } catch (JMSException e) {

     e.printStackTrace();
    } finally {
                 try {
      session.commit();
      session.close();
      connection.close();
     } catch (JMSException e) {
 
      e.printStackTrace();
     }
    }
               
    
   }
  });
  
  
  
  
 }

}