<< 深入掌握JMS 12 系列 | 首页 | Synchronous Request Response with ActiveMQ and Spring | CodeDependents >>

JMS&MQ系列之JMS的请求和回应 - geloin - 博客频道 - CSDN.NET

代理类：

 

[java] view plaincopy
 
  1. /** 
  2.  *  
  3.  * @author geloin 
  4.  * @date 2012-9-14 下午5:57:36 
  5.  */  
  6. package com.geloin.activemq.test4;  
  7.   
  8. import org.apache.activemq.broker.BrokerService;  
  9.   
  10. /** 
  11.  *  
  12.  * @author geloin 
  13.  * @date 2012-9-14 下午5:57:36 
  14.  */  
  15. public class Broker {  
  16.       
  17.     /** 
  18.      * 创建并启动代理 
  19.      *  
  20.      * @author geloin 
  21.      * @date 2012-9-14 下午5:58:35 
  22.      * @throws Exception 
  23.      */  
  24.     private void createBroker() throws Exception {  
  25.         BrokerService broker = new BrokerService();  
  26.         broker.setPersistent(false);  
  27.         broker.setUseJmx(false);  
  28.         broker.addConnector("tcp://localhost:61616");  
  29.         broker.start();  
  30.           
  31.         System.out.println("\nPress any key to stop broker\n");  
  32.         System.in.read();  
  33.           
  34.         broker.start();  
  35.     }  
  36.       
  37.     /** 
  38.      *  
  39.      *  
  40.      * @author geloin 
  41.      * @date 2012-9-14 下午5:59:30 
  42.      * @param args 
  43.      * @throws Exception 
  44.      */  
  45.     public static void main(String[] args) throws Exception {  
  46.         Broker broker = new Broker();  
  47.         broker.createBroker();  
  48.     }  
  49. }  


        服务端：

 

 

[java] view plaincopy
 
  1. /** 
  2.  *  
  3.  * @author geloin 
  4.  * @date 2012-9-14 下午5:37:38 
  5.  */  
  6. package com.geloin.activemq.test4;  
  7.   
  8. import javax.jms.Connection;  
  9. import javax.jms.DeliveryMode;  
  10. import javax.jms.Destination;  
  11. import javax.jms.Message;  
  12. import javax.jms.MessageConsumer;  
  13. import javax.jms.MessageListener;  
  14. import javax.jms.MessageProducer;  
  15. import javax.jms.Session;  
  16. import javax.jms.TextMessage;  
  17.   
  18. import org.apache.activemq.ActiveMQConnectionFactory;  
  19.   
  20. /** 
  21.  * 服务进程 
  22.  *  
  23.  * @author geloin 
  24.  * @date 2012-9-14 下午5:37:38 
  25.  */  
  26. public class Server implements MessageListener {  
  27.   
  28.     private String brokerURL = "tcp://localhost:61616";  
  29.     private Connection conn;  
  30.     private Session session;  
  31.     private String requestQueue = "TEST.QUEUE";  
  32.     private MessageProducer producer;  
  33.     private MessageConsumer consumer;  
  34.   
  35.     /** 
  36.      * 消息处理 
  37.      *  
  38.      * @author geloin 
  39.      * @date 2012-9-14 下午5:53:46 
  40.      * @param messageText 
  41.      * @return 
  42.      */  
  43.     private String handleRequest(String messageText) {  
  44.         return "Response to '" + messageText + "'";  
  45.     }  
  46.   
  47.     /* 
  48.      * (non-Javadoc) 
  49.      *  
  50.      * @see javax.jms.MessageListener#onMessage(javax.jms.Message) 
  51.      */  
  52.     @Override  
  53.     public void onMessage(Message message) {  
  54.         // 监听到有消息传送至此时，执行onMessage  
  55.         try {  
  56.             // 若有消息传送到服务时，先创建一个文本消息  
  57.             TextMessage response = this.session.createTextMessage();  
  58.             // 若从客户端传送到服务端的消息为文本消息  
  59.             if (message instanceof TextMessage) {  
  60.                 // 先将传送到服务端的消息转化为文本消息  
  61.                 TextMessage txtMsg = (TextMessage) message;  
  62.                 // 取得文本消息的内容  
  63.                 String messageText = txtMsg.getText();  
  64.                 // 将客户端传送过来的文本消息进行处理后，设置到回应消息里面  
  65.                 response.setText(handleRequest(messageText));  
  66.             }  
  67.             // 设置回应消息的关联ID，关联ID来自于客户端传送过来的关联ID  
  68.             response.setJMSCorrelationID(message.getJMSCorrelationID());  
  69.             // 生产者发送回应消息，目的由客户端的JMSReplyTo定义，内容即刚刚定义的回应消息  
  70.             producer.send(message.getJMSReplyTo(), response);  
  71.   
  72.         } catch (Exception e) {  
  73.             e.printStackTrace();  
  74.         }  
  75.     }  
  76.   
  77.     /** 
  78.      *  
  79.      * @author geloin 
  80.      * @date 2012-9-14 下午5:37:39 
  81.      * @param args 
  82.      */  
  83.     public static void main(String[] args) throws Exception {  
  84.         // 定义并启动服务  
  85.         Server server = new Server();  
  86.         server.start();  
  87.   
  88.         // 监听控制器输入  
  89.         System.out.println("\nPress any key to stop the server\n");  
  90.         System.in.read();  
  91.   
  92.         // 停止服务  
  93.         server.stop();  
  94.     }  
  95.   
  96.     /** 
  97.      * 服务 
  98.      *  
  99.      * @author geloin 
  100.      * @date 2012-9-14 下午5:55:49 
  101.      * @throws Exception 
  102.      */  
  103.     public void start() throws Exception {  
  104.         // 使用tcp协议创建连接通道并启动  
  105.         ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(  
  106.                 brokerURL);  
  107.         conn = factory.createConnection();  
  108.         conn.start();  
  109.   
  110.         // 创建session及消息的目的地，并设定交互时使用的存储方式，同时定义队列名称，客户端通过此名称连接  
  111.         session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);  
  112.         Destination dest = session.createQueue(requestQueue);  
  113.   
  114.         // 创建生产者，并设定分配模式，生产者的目的地为null，因为它的目的地由JMSReplyTo定义  
  115.         producer = session.createProducer(null);  
  116.         producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);  
  117.   
  118.         // 消费者，及消费者的临听程序  
  119.         consumer = session.createConsumer(dest);  
  120.         consumer.setMessageListener(this);  
  121.     }  
  122.   
  123.     /** 
  124.      * 回收资源 
  125.      *  
  126.      * @author geloin 
  127.      * @date 2012-9-13 上午9:42:06 
  128.      * @throws Exception 
  129.      */  
  130.     private void stop() {  
  131.         try {  
  132.             if (null != producer) {  
  133.                 producer.close();  
  134.             }  
  135.   
  136.             if (null != consumer) {  
  137.                 consumer.close();  
  138.             }  
  139.   
  140.             if (null != session) {  
  141.                 session.close();  
  142.             }  
  143.   
  144.             if (null != conn) {  
  145.                 conn.close();  
  146.             }  
  147.         } catch (Exception e) {  
  148.             e.printStackTrace();  
  149.         }  
  150.     }  
  151. }  


        客户端：

 

 

[java] view plaincopy
 
  1. /** 
  2.  *  
  3.  * @author geloin 
  4.  * @date 2012-9-14 下午6:00:15 
  5.  */  
  6. package com.geloin.activemq.test4;  
  7.   
  8. import java.util.UUID;  
  9.   
  10. import javax.jms.Connection;  
  11. import javax.jms.DeliveryMode;  
  12. import javax.jms.Destination;  
  13. import javax.jms.JMSException;  
  14. import javax.jms.Message;  
  15. import javax.jms.MessageConsumer;  
  16. import javax.jms.MessageListener;  
  17. import javax.jms.MessageProducer;  
  18. import javax.jms.Session;  
  19. import javax.jms.TextMessage;  
  20.   
  21. import org.apache.activemq.ActiveMQConnectionFactory;  
  22.   
  23. /** 
  24.  *  
  25.  * @author geloin 
  26.  * @date 2012-9-14 下午6:00:15 
  27.  */  
  28. public class Client implements MessageListener {  
  29.   
  30.     private String brokerURL = "tcp://localhost:61616";  
  31.     private Connection conn;  
  32.     private Session session;  
  33.     private String requestQueue = "TEST.QUEUE";  
  34.     private MessageProducer producer;  
  35.     private MessageConsumer consumer;  
  36.     private Destination tempDest;  
  37.       
  38.     /* (non-Javadoc) 
  39.      * @see javax.jms.MessageListener#onMessage(javax.jms.Message) 
  40.      */  
  41.     @Override  
  42.     public void onMessage(Message message) {  
  43.         try {  
  44.             System.out.println("Receive response for: "  
  45.                     + ((TextMessage) message).getText());  
  46.         } catch (JMSException e) {  
  47.             e.printStackTrace();  
  48.         }  
  49.     }  
  50.   
  51.     /** 
  52.      * 启动 
  53.      *  
  54.      * @author geloin 
  55.      * @date 2012-9-13 下午7:49:00 
  56.      */  
  57.     private void start() {  
  58.         try {  
  59.             // 使用tcp协议创建连接通道并启动  
  60.             ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(  
  61.                     brokerURL);  
  62.             conn = factory.createConnection();  
  63.             conn.start();  
  64.               
  65.             // 创建session及消息的目的地，并设定交互时使用的存储方式，同时定义队列名称，此名称与服务端相同  
  66.             session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);  
  67.             Destination dest = session.createQueue(requestQueue);  
  68.   
  69.             // 创建生产者，并设定分配模式  
  70.             producer = session.createProducer(dest);  
  71.             producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);  
  72.   
  73.             // 使用临时目的地创建消费者，及消费者的临听程序  
  74.             tempDest = session.createTemporaryQueue();  
  75.             consumer = session.createConsumer(tempDest);  
  76.             consumer.setMessageListener(this);  
  77.         } catch (Exception e) {  
  78.             e.printStackTrace();  
  79.         }  
  80.     }  
  81.   
  82.     /** 
  83.      * 停止 
  84.      *  
  85.      * @author geloin 
  86.      * @date 2012-9-13 下午7:49:06 
  87.      */  
  88.     public void stop() {  
  89.         try {  
  90.             if (null != producer) {  
  91.                 producer.close();  
  92.             }  
  93.   
  94.             if (null != consumer) {  
  95.                 consumer.close();  
  96.             }  
  97.   
  98.             if (null != session) {  
  99.                 session.close();  
  100.             }  
  101.   
  102.             if (null != conn) {  
  103.                 conn.close();  
  104.             }  
  105.         } catch (Exception e) {  
  106.             e.printStackTrace();  
  107.         }  
  108.     }  
  109.   
  110.     /** 
  111.      * 请求 
  112.      *  
  113.      * @author geloin 
  114.      * @date 2012-9-13 下午7:52:54 
  115.      * @param request 
  116.      * @throws Exception 
  117.      */  
  118.     public void request(String request) throws Exception {  
  119.         System.out.println("Requesting:" + request);  
  120.         TextMessage txtMsg = session.createTextMessage();  
  121.         txtMsg.setText(request);  
  122.   
  123.         txtMsg.setJMSReplyTo(tempDest);  
  124.   
  125.         String correlationId = UUID.randomUUID().toString();  
  126.         txtMsg.setJMSCorrelationID(correlationId);  
  127.   
  128.         producer.send(txtMsg);  
  129.     }  
  130.       
  131.     /** 
  132.      *  
  133.      * @author geloin 
  134.      * @date 2012-9-14 下午6:00:15 
  135.      * @param args 
  136.      */  
  137.     public static void main(String[] args) throws Exception {  
  138.         Client client = new Client();  
  139.         client.start();  
  140.         int i = 0;  
  141.         while (i++ < 10) {  
  142.             client.request("REQUEST-" + i);  
  143.         }  
  144.         Thread.sleep(3000);  
  145.         client.stop();  
  146.     }  
  147.   
  148. }  


        代码比较简单，是实现JMS的基本过程，从以上代码中，我们可以得出以下结论：

 

        1. 要实现交互，必须有代理，即Broker，该代理必须设定Connector，也即是我们所谓brokerURL。在服务端与客户端交互过程中，broker必须被启动；

        2. 要向外发送消息，需要执行以下步骤：

        (1) 通过第一步中的brokerURL创建ConnectionFactory；

        (2) 通过ConnectionFactory创建Connection并启动；

        (3) 通过Connection创建Session；

        (4) 通过Session创建发送目标；

        (5) 通过Session创建MessageProducer；

        (6) 通过Session创建Message；

        (7) 通过MessageProducer发送Message。

        3. 要接收消息，需要执行以下步骤：

 

        (1) 通过第一步中的brokerURL创建ConnectionFactory；

        (2) 通过ConnectionFactory创建Connection并启动；

        (3) 通过Connection创建Session；

        (4) 通过Session创建发送目标；

        (5) 通过Session创建MessageConsumer；

        (6) 通过MessageConsumer取得Message并分析处理。

        4. 若要实现交互，则注意以下几点：

        (1) 服务端和客户端使用同一个brokerURL；

        (2) 通常情况下，服务端和客户端各有一个MessageProducer和MessageConsumer；

        (3) 为使服务端能够回应多个客户端，通常将其MessageProducer的Destination设置为null，即不设定，而由JMSReplyTo定义；

        (4) 对应于服务端的MessageProducer的为null的Destination，若确定服务端与客户端能够交互，则在客户端可设置其MessageConsumer的Destination为临时Destination；

        (5) 为使服务端能够正常回应客户端，客户端需设置消息的JMSReplyTo属性及JMSCorrelationID，服务端需设置消息的JMSCorrelationID为客户端设定的JMSCorrelationID，producer的send的Destination为客户端设定的JMSReplyTo。

        5. 两个需要交互的MessageProducer和MessageConsumer之间除需要使用同一brokerURL外，还需要保障其Destination的对应，即保持在创建Destination时使用的queueName相同。

阅读全文……

标签 : ,
发表评论 RSS feed for responses to this blog entry
IT瘾于2014年8月6日 上午01时40分00秒发布 #



发表评论 发送引用通报