ActiveMQ接收消息+發送消息的簡單實例

jopen 11年前發布 | 43K 次閱讀 ActiveMQ 消息系統

最近由于公司項目需要 -ActiveMQ接收+發送消息，用的是ActiveMQ</span>。由于這方面網上的例子不是很多，而且有的也不完整。于是經過幾天的摸索學習，在網上找到了合適的方案。

 

ProducerTool.java用于發送消息：

java 代碼
  1. package homework;   
  2.   
  3. import javax.jms.Connection;   
  4. import javax.jms.DeliveryMode;   
  5. import javax.jms.Destination;   
  6. import javax.jms.JMSException;   
  7. import javax.jms.MessageProducer;   
  8. import javax.jms.Session;   
  9. import javax.jms.TextMessage;   
  10.   
  11. import org.apache.activemq.ActiveMQConnection;   
  12. import org.apache.activemq.ActiveMQConnectionFactory;   
  13.   
  14. public class ProducerTool {   
  15.   
  16.     private String user = ActiveMQConnection.DEFAULT_USER;   
  17.   
  18.     private String password = ActiveMQConnection.DEFAULT_PASSWORD;   
  19.   
  20.     private String url = ActiveMQConnection.DEFAULT_BROKER_URL;   
  21.   
  22.     private String subject = "TOOL.DEFAULT";   
  23.   
  24.     private Destination destination = null;   
  25.   
  26.     private Connection connection = null;   
  27.   
  28.     private Session session = null;   
  29.   
  30.     private MessageProducer producer = null;   
  31.   
  32.     // 初始化   
  33.     private void initialize() throws JMSException, Exception {   
  34.         ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(   
  35.                 user, password, url);   
  36.         connection = connectionFactory.createConnection();   
  37.         session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);   
  38.         destination = session.createQueue(subject);   
  39.         producer = session.createProducer(destination);   
  40.         producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);   
  41.     }   
  42.   
  43.     // 發送消息  
  44.     public void produceMessage(String message) throws JMSException, Exception {   
  45.         initialize();   
  46.         TextMessage msg = session.createTextMessage(message);   
  47.         connection.start();   
  48.         System.out.println("Producer:->Sending message: " + message);   
  49.         producer.send(msg);   
  50.         System.out.println("Producer:->Message sent complete!");   
  51.     }   
  52.   
  53.     // 關閉連接  
  54.     public void close() throws JMSException {   
  55.         System.out.println("Producer:->Closing connection");   
  56.         if (producer != null)   
  57.             producer.close();   
  58.         if (session != null)   
  59.             session.close();   
  60.         if (connection != null)   
  61.             connection.close();   
  62.     }   
  63. }   

 

ConsumerTool.java用于接受消息，我用的是基于消息監聽的機制，需要實現MessageListener接口，這個接口有個onMessage方法，當接受到消息的時候會自動調用這個函數對消息進行處理。

java 代碼
  1. package homework;   
  2.   
  3. import javax.jms.Connection;   
  4. import javax.jms.Destination;   
  5. import javax.jms.JMSException;   
  6. import javax.jms.MessageConsumer;   
  7. import javax.jms.Session;   
  8. import javax.jms.MessageListener;   
  9. import javax.jms.Message;   
  10. import javax.jms.TextMessage;   
  11.   
  12. import org.apache.activemq.ActiveMQConnection;   
  13. import org.apache.activemq.ActiveMQConnectionFactory;   
  14.   
  15. public class ConsumerTool implements MessageListener {   
  16.   
  17.     private String user = ActiveMQConnection.DEFAULT_USER;   
  18.   
  19.     private String password = ActiveMQConnection.DEFAULT_PASSWORD;   
  20.   
  21.     private String url = ActiveMQConnection.DEFAULT_BROKER_URL;   
  22.   
  23.     private String subject = "TOOL.DEFAULT";   
  24.   
  25.     private Destination destination = null;   
  26.   
  27.     private Connection connection = null;   
  28.   
  29.     private Session session = null;   
  30.   
  31.     private MessageConsumer consumer = null;   
  32.   
  33.     // 初始化  
  34.     private void initialize() throws JMSException, Exception {   
  35.         ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(   
  36.                 user, password, url);   
  37.         connection = connectionFactory.createConnection();   
  38.         session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);   
  39.         destination = session.createQueue(subject);   
  40.         consumer = session.createConsumer(destination);   
  41.            
  42.     }   
  43.   
  44.     // 消費消息   
  45.     public void consumeMessage() throws JMSException, Exception {   
  46.         initialize();   
  47.         connection.start();   
  48.            
  49.         System.out.println("Consumer:->Begin listening...");   
  50.         // 開始監聽  
  51.         consumer.setMessageListener(this);   
  52.         // Message message = consumer.receive();  
  53.     }   
  54.   
  55.     // 關閉連接  
  56.     public void close() throws JMSException {   
  57.         System.out.println("Consumer:->Closing connection");   
  58.         if (consumer != null)   
  59.             consumer.close();   
  60.         if (session != null)   
  61.             session.close();   
  62.         if (connection != null)   
  63.             connection.close();   
  64.     }   
  65.   
  66.     // 消息處理函數   
  67.     public void onMessage(Message message) {   
  68.         try {   
  69.             if (message instanceof TextMessage) {   
  70.                 TextMessage txtMsg = (TextMessage) message;   
  71.                 String msg = txtMsg.getText();   
  72.                 System.out.println("Consumer:->Received: " + msg);   
  73.             } else {   
  74.                 System.out.println("Consumer:->Received: " + message);   
  75.             }   
  76.         } catch (JMSException e) {   
  77.             // TODO Auto-generated catch block  
  78.             e.printStackTrace();   
  79.         }   
  80.     }   
  81. }   

 

如果想主動的去接受消息，而不用消息監聽的話，把consumer.setMessageListener(this)改為Message message = consumer.receive()，手動去調用MessageConsumer的receive方法即可。

下面是測試類Test.java：

java 代碼
  1. package homework;   
  2.   
  3. import javax.jms.JMSException;   
  4.   
  5. public class Test {   
  6.   
  7.     /** 
  8.      * @param args  
  9.      */  
  10.     public static void main(String[] args) throws JMSException, Exception {   
  11.         // TODO Auto-generated method stub  
  12.         ConsumerTool consumer = new ConsumerTool();   
  13.         ProducerTool producer = new ProducerTool();   
  14.         // 開始監聽  
  15.         consumer.consumeMessage();   
  16.            
  17.         // 延時500毫秒之后發送消息  
  18.         Thread.sleep(500);   
  19.         producer.produceMessage("Hello, world!");   
  20.         producer.close();   
  21.            
  22.         // 延時500毫秒之后停止接受消息  
  23.         Thread.sleep(500);   
  24.         consumer.close();   
  25.     }   
  26. }   

 

ActiveMQ的一個簡單實例-ActiveMQ接收+發送消息

 本文由用戶 jopen 自行上傳分享，僅供網友學習交流。所有權歸原作者，若您的權利被侵害，請聯系管理員。
 轉載本站原創文章，請注明出處，并保留原始鏈接、圖片水印。
 本站是一個以用戶分享為主的開源技術平臺，歡迎各類分享！