ActiveMQ的一个简单示例 ActiveMQJMSApachethread

最近由于公司项目需要，开始学习JMS，用的是ActiveMQ。由于这方面网上的例子不是很多，而且有的也不完整。于是经过几天的摸索学习，写了一个简单的小例子，现在贴出来与大家分享。
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之后写的一个简单的例子，希望对你有帮助，如果有什么错误还请指正。