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/