ActiveMQ接收消息+发送消息的简单实例

jopen 12年前

最近由于公司项目需要 -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接收+发送消息