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

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

ActiveMQ的一个简单实例-ActiveMQ接收+发送消息   JMS消息框架——ActiveMQ

最近由于公司项目需要 -ActiveMQ接收+发送消息，用的是ActiveMQ。由于这方面网上的例子不是很多，而且有的也不完整。于是经过几天的摸索学习，在网上找到了合适的方案。

我的 IT技术资源库   http://itlib.tk/

ProducerTool.java用于发送消息：

java 代码

1. package homework;     
2. import javax.jms.Connection;   import javax.jms.DeliveryMode;   
3. import javax.jms.Destination;   import javax.jms.JMSException;   
4. import javax.jms.MessageProducer;   import javax.jms.Session;   
5. import javax.jms.TextMessage;     
6. import org.apache.activemq.ActiveMQConnection;    import org.apache.activemq.ActiveMQConnectionFactory;   
7.    public class ProducerTool {   
8.        private String user = ActiveMQConnection.DEFAULT_USER;   
9.        private String password = ActiveMQConnection.DEFAULT_PASSWORD;   
10.        private String url = ActiveMQConnection.DEFAULT_BROKER_URL;   
11.        private String subject = "TOOL.DEFAULT";   
12.        private Destination destination = null;   
13.        private Connection connection = null;   
14.        private Session session = null;   
15.        private MessageProducer producer = null;   
16.        // 初始化   
17.     private void initialize() throws JMSException, Exception {            ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(   
18.                 user, password, url);            connection = connectionFactory.createConnection();   
19.         session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);           destination = session.createQueue(subject);   
20.         producer = session.createProducer(destination);           producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);   
21.     }      
22.     // 发送消息      public void produceMessage(String message) throws JMSException, Exception {   
23.         initialize();            TextMessage msg = session.createTextMessage(message);   
24.         connection.start();            System.out.println("Producer:->Sending message: " + message);   
25.         producer.send(msg);            System.out.println("Producer:->Message sent complete!");   
26.     }      
27.     // 关闭连接      public void close() throws JMSException {   
28.         System.out.println("Producer:->Closing connection");           if (producer != null)   
29.             producer.close();            if (session != null)   
30.             session.close();            if (connection != null)   
31.             connection.close();        }   
32. }   

 

ConsumerTool.java用于接受消息，我用的是基于消息监听的机制，需要实现MessageListener接口，这个接口有个onMessage方法，当接受到消息的时候会自动调用这个函数对消息进行处理。

java 代码

1. package homework;     
2. import javax.jms.Connection;   import javax.jms.Destination;   
3. import javax.jms.JMSException;   import javax.jms.MessageConsumer;   
4. import javax.jms.Session;   import javax.jms.MessageListener;   
5. import javax.jms.Message;   import javax.jms.TextMessage;   
6.    import org.apache.activemq.ActiveMQConnection;   
7. import org.apache.activemq.ActiveMQConnectionFactory;      
8. public class ConsumerTool implements MessageListener {      
9.     private String user = ActiveMQConnection.DEFAULT_USER;     
10.     private String password = ActiveMQConnection.DEFAULT_PASSWORD;      
11.     private String url = ActiveMQConnection.DEFAULT_BROKER_URL;     
12.     private String subject = "TOOL.DEFAULT";     
13.     private Destination destination = null;     
14.     private Connection connection = null;     
15.     private Session session = null;     
16.     private MessageConsumer consumer = null;     
17.     // 初始化      private void initialize() throws JMSException, Exception {   
18.         ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(                    user, password, url);   
19.         connection = connectionFactory.createConnection();           session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);   
20.         destination = session.createQueue(subject);           consumer = session.createConsumer(destination);   
21.                 }   
22.        // 消费消息   
23.     public void consumeMessage() throws JMSException, Exception {            initialize();   
24.         connection.start();               
25.         System.out.println("Consumer:->Begin listening...");           // 开始监听  
26.         consumer.setMessageListener(this);           // Message message = consumer.receive();  
27.     }      
28.     // 关闭连接      public void close() throws JMSException {   
29.         System.out.println("Consumer:->Closing connection");           if (consumer != null)   
30.             consumer.close();            if (session != null)   
31.             session.close();            if (connection != null)   
32.             connection.close();        }   
33.        // 消息处理函数   
34.     public void onMessage(Message message) {           try {   
35.             if (message instanceof TextMessage) {                    TextMessage txtMsg = (TextMessage) message;   
36.                 String msg = txtMsg.getText();                    System.out.println("Consumer:->Received: " + msg);   
37.             } else {                   System.out.println("Consumer:->Received: " + message);   
38.             }            } catch (JMSException e) {   
39.             // TODO Auto-generated catch block              e.printStackTrace();   
40.         }        }   
41. }   

 

如果想主动的去接受消息，而不用消息监听的话，把consumer.setMessageListener(this)改为Message message = consumer.receive()，手动去调用MessageConsumer的receive方法即可。

下面是测试类Test.java：

java 代码

1. package homework;     
2. import javax.jms.JMSException;     
3. public class Test {     
4.     /**      * @param args  
5.      */      public static void main(String[] args) throws JMSException, Exception {   
6.         // TODO Auto-generated method stub          ConsumerTool consumer = new ConsumerTool();   
7.         ProducerTool producer = new ProducerTool();           // 开始监听  
8.         consumer.consumeMessage();               
9.         // 延时500毫秒之后发送消息          Thread.sleep(500);   
10.         producer.produceMessage("Hello, world!");           producer.close();   
11.                     // 延时500毫秒之后停止接受消息  
12.         Thread.sleep(500);           consumer.close();   
13.     }    }   

 

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

IT技术资源库   http://itlib.tk/