Active MQ (二) Java Application编程
1.准备工作
? 1）启动ActiveMQ
? 2）将%ActiveMQ_HOME%\activemq-all-5.1.0.jar导入对应的JAVA Project
?
2.对于MQ编程通常使用JMS编程，但是由于纯JAVA Application Project中不存在JNDI容器，所以还是在
? 创建ConnectionFactory时还是不得不与ActiveMQ API耦合
?
3.JMS体系结构
?? 
?
JMS有两种消息模式：
1）一对一/点到点 模式
2）预订/发布 模式
JMS编程主要的接口有：?
ConnectionFactory
Connection ： 一般一个aplication中只有一个，线程安全，可复用。
Session
Queue
Topic
MessageProducer
MessageConsumer
MessageListener
?
4.实例
1）一对一/点到点
?
QueueReceiverListener.java?
?
package com.siyuan.jms;
import java.util.Date;
import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;
public class QueueReceiverListener implements MessageListener {
?
?private Connection connection;
?
?private String queueNameSendTo;
?
?/**
? * @return the connection
? */
?public Connection getConnection() {
??return connection;
?}
?/**
? * @param connection the connection to set
? */
?public void setConnection(Connection connection) {
??this.connection = connection;
?}
?/**
? * @return the queueNameSendTo
? */
?public String getQueueNameSendTo() {
??return queueNameSendTo;
?}
?/**
? * @param queueNameSendTo the queueNameSendTo to set
? */
?public void setQueueNameSendTo(String queueNameSendTo) {
??this.queueNameSendTo = queueNameSendTo;
?}
?public void onMessage(Message message) {
??if (message instanceof TextMessage) {
???try {
????System.out.println("Current time : " + new Date());
????System.out.println("Receive : " + ((TextMessage) message).getText());
????System.out.println("Receive JMSCorrelationID : " +? message.getJMSCorrelationID());
????
????Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
????Queue queueSendTo = session.createQueue(queueNameSendTo);
????MessageProducer sender = session.createProducer(queueSendTo);
????TextMessage messageToSend = session.createTextMessage();
????messageToSend.setText("Nice to meet u, too.");
????messageToSend.setJMSCorrelationID(message.getJMSCorrelationID());
????sender.send(messageToSend);
????System.out.println("Send : " + messageToSend.getText());
????System.out.println("Send JMSCorrelationID : " +? messageToSend.getJMSCorrelationID());
????System.out.println("--------------------------------------------");
????sender.close();
????session.close();
???} catch (JMSException e) {
????// TODO Auto-generated catch block
????e.printStackTrace();
???}
??}
?}
}
====================================================================
?
JMSQueueTest.java
?
package com.siyuan.jms;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.Queue;
import javax.jms.Session;
import org.apache.activemq.ActiveMQConnectionFactory;
public class JMSQueueTest {
?
?public static String QUEUE_RECEIVE_FROM_NAME = "MY.TEST.SENDER";
?public static String QUEUE_SEND_TO_NAME = "MY.TEST.RECEIVER";
?
?/**
? * @param args
? * @throws JMSException
? */
?public static void main(String[] args) throws JMSException {
??
??ConnectionFactory connFactory
???= new ActiveMQConnectionFactory("tcp://localhost:61616");
??Connection connection = connFactory.createConnection();
??connection.start();
??
??Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
??
??Queue queue = session.createQueue(QUEUE_RECEIVE_FROM_NAME);
??MessageConsumer messageProcessor = session.createConsumer(queue);
??QueueReceiverListener processor = new QueueReceiverListener();
??processor.setConnection(connection);
??processor.setQueueNameSendTo(QUEUE_SEND_TO_NAME);
??messageProcessor.setMessageListener(processor);
??
?}
}
?
说明：本例为模拟一消息接收处理器，即从一Queue中获取Req消息，处理后将
???????? Resp消息发送至另一Queue中，其中Req和Resp通过JMSCorrelationID
???????? 进行关联。相关的Req消息发送器将在ActiveMQ和Tomcat集成中在JSP中实现。
编程中需注意：（1） connection.start();
???????????????????? （2） 异步方式获取消息，不能connection.close();
?
2）预订/发布
?
JMSTopicTest.java
?
package com.siyuan.jms;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;
import org.apache.activemq.ActiveMQConnectionFactory;
public class JMSTopicTest {
?/**
? * @param args
? * @throws JMSException
? */
?public static void main(String[] args) throws JMSException {
??// TODO Auto-generated method stub
??ConnectionFactory connFactory
???= new ActiveMQConnectionFactory("tcp://localhost:61616");
??Connection connection = connFactory.createConnection();
??connection.start();
??
??Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
??
??Topic topic = session.createTopic("MY.TEST.TOPIC");
? MessageConsumer subscriber = session.createConsumer(topic);
??MessageProducer publisher = session.createProducer(topic);
??publisher.setDeliveryMode(DeliveryMode.NON_PERSISTENT);???
??TextMessage messageToPublish = session.createTextMessage();
??messageToPublish.setText("Hello all!");
??publisher.send(messageToPublish);
??
????Message messageFromSubscribe = subscriber.receive(3000);
??System.out.println(messageFromSubscribe);
??if (messageFromSubscribe instanceof TextMessage) {
???System.out.println(((TextMessage) messageFromSubscribe).getText());
??}
??publisher.close();
??subscriber.close();
??session.close();
??connection.close();
?}
}
?
编程中需注意：（1） subscriber定义必须在publisher定义之前
?????????????????????????????? 只能预订后才能收到发布的消息
?
?5.参考资料：
http://orange5458.iteye.com/admin/blogs/991311
JMS学习文档 PDF格式
?
