使用 ActiveMQ 示例

企业中各项目中相互协作的时候可能用得到消息通知机制。比如有东西更新了，可以通知做索引。

在 Java 里有 JMS 的多个实现。其中 apache 下的 ActiveMQ 就是不错的选择。还有一个比较热的是 RabbitMQ （是 erlang 语言实现的）。这里示例下使用 ActiveMQ
用 ActiveMQ 最好还是了解下 JMS
JMS 公共 点对点域 发布/订阅域
ConnectionFactory QueueConnectionFactory TopicConnectionFactory
Connection QueueConnection TopicConnection
Destination Queue Topic
Session QueueSession TopicSession
MessageProducer QueueSender TopicPublisher
MessageConsumer QueueReceiver TopicSubscriber



JMS 定义了两种方式：Quere（点对点）；Topic（发布/订阅）。
ConnectionFactory 是连接工厂，负责创建Connection。
Connection 负责创建 Session。
Session 创建 MessageProducer（用来发消息） 和 MessageConsumer（用来接收消息）。
Destination 是消息的目的地。
详细的可以网上找些 JMS 规范（有中文版）。
下载 apache-activemq-5.3.0。http://activemq.apache.org/download.html， 解压，然后双击 bin/activemq.bat。运行后，可以在 http://localhost:8161/admin 观察。也有 demo， http://localhost:8161/demo。 把 activemq-all-5.3.0.jar 加入 classpath。

Jms 发送 代码：

package point.to.point;

import java.util.Date;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.MessageProducer;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnectionFactory;

public class JMSSend {

/**
* @param args
* @throws JMSException
*/
public static void main(String[] args) throws JMSException {
// TODO Auto-generated method stub
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory();

Connection connection = factory.createConnection();
connection.start();
Session session = connection.createSession(true,
Session.AUTO_ACKNOWLEDGE);
Destination des = session.createQueue("myquen-1");
MessageProducer pro = session.createProducer(des);
MapMessage msg = session.createMapMessage();
msg.setString("time", new Date().toLocaleString());
pro.send(msg);
session.commit();
session.close();
connection.close();
}

}

jms接受代码

package point.to.point;

import java.util.Date;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;

import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;

import javax.jms.Session;

import org.apache.activemq.ActiveMQConnectionFactory;

public class JMSReceiver {
static boolean istrue=true;

/**
* @param args
* @throws JMSException
*/
public static void main(String[] args) throws JMSException {
// TODO Auto-generated method stub
ConnectionFactory factory = new ActiveMQConnectionFactory();
Connection connection = factory.createConnection();
connection.start();
Session session = connection.createSession(true,
Session.AUTO_ACKNOWLEDGE);
Destination des = session.createQueue("myquen-1");
MessageConsumer pro = session.createConsumer(des);
while(istrue){
MapMessage msg =(MapMessage)pro.receive();
System.out.println(msg.getString("time"));
//接收时必须提交以下 否则会出现上次收到的旧数据
session.commit();
}
session.close();
connection.close();
}

}

结合 Spring 的 JmsTemplate 方便用：
xml01.
<?xml version="1.0" encoding="UTF-8"?> 
<beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans [url]http://www.springframework.org/schema/beans/spring-beans-2.5.xsd[/url]"> 

<!-- 在非 web / ejb 容器中使用 pool 时，要手动 stop，spring 不会为你执行 destroy-method 的方法

<bean id="jmsFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop">

<property name="connectionFactory">

<bean class="org.apache.activemq.ActiveMQConnectionFactory">

<property name="brokerURL" value="tcp://localhost:61616" />

</bean>

</property>

</bean>

-->

<bean id="jmsFactory" class="org.apache.activemq.ActiveMQConnectionFactory">

<property name="brokerURL" value="tcp://localhost:61616" />

</bean>

<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">

<property name="connectionFactory" ref="jmsFactory" />

<property name="defaultDestination" ref="destination" />

<property name="messageConverter">

<bean class="org.springframework.jms.support.converter.SimpleMessageConverter" />

</property>

</bean> 
<bean id="destination" class="org.apache.activemq.command.ActiveMQQueue">

<constructor-arg index="0" value="my-queue" />

</bean>

</beans>
<?xml version="1.0" encoding="UTF-8"?> 

<beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

xsi:schemaLocation="http://www.springframework.org/schema/beans [url]http://www.springframework.org/schema/beans/spring-beans-2.5.xsd[/url]"> 
<!-- 在非 web / ejb 容器中使用 pool 时，要手动 stop，spring 不会为你执行 destroy-method 的方法

<bean id="jmsFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop">

<property name="connectionFactory">

<bean class="org.apache.activemq.ActiveMQConnectionFactory"> 
<property name="brokerURL" value="tcp://localhost:61616" /> 
</bean> 
</property>

</bean> 
--> 
<bean id="jmsFactory" class="org.apache.activemq.ActiveMQConnectionFactory"> 
<property name="brokerURL" value="tcp://localhost:61616" /> 
</bean> 
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate"> 
<property name="connectionFactory" ref="jmsFactory" /> 
<property name="defaultDestination" ref="destination" /> 
<property name="messageConverter"> 
<bean class="org.springframework.jms.support.converter.SimpleMessageConverter" /> 
</property> 
</bean> 
<bean id="destination" class="org.apache.activemq.command.ActiveMQQueue">

<constructor-arg index="0" value="my-queue" />

</bean> 
</beans>
sender：01.
public static void main(String[] args) {

ApplicationContext ctx = new FileSystemXmlApplicationContext("classpath:app*.xml");

JmsTemplate jmsTemplate = (JmsTemplate) ctx.getBean("jmsTemplate"); 
jmsTemplate.send(new MessageCreator() { 
public Message createMessage(Session session) throws JMSException { 
MapMessage mm = session.createMapMessage(); 
mm.setLong("count", new Date().getTime()); 
return mm; 
} 
}); 
}
public static void main(String[] args) { 
ApplicationContext ctx = new FileSystemXmlApplicationContext("classpath:app*.xml"); 
JmsTemplate jmsTemplate = (JmsTemplate) ctx.getBean("jmsTemplate"); 
jmsTemplate.send(new MessageCreator() { 
public Message createMessage(Session session) throws JMSException { 
MapMessage mm = session.createMapMessage(); 
mm.setLong("count", new Date().getTime()); 
return mm;

} 
});

}
receiver：01.
public static void main(String[] args) {

ApplicationContext ctx = new FileSystemXmlApplicationContext("classpath:app*.xml"); 
JmsTemplate jmsTemplate = (JmsTemplate) ctx.getBean("jmsTemplate");

while(true) {

Map<String, Object> mm = (Map<String, Object>) jmsTemplate.receiveAndConvert();

System.out.println("收到消息：" + new Date((Long)mm.get("count"))); 
}

}

public static void main(String[] args) { 
ApplicationContext ctx = new FileSystemXmlApplicationContext("classpath:app*.xml"); 
JmsTemplate jmsTemplate = (JmsTemplate) ctx.getBean("jmsTemplate"); 
while(true) {

Map<String, Object> mm = (Map<String, Object>) jmsTemplate.receiveAndConvert();

System.out.println("收到消息：" + new Date((Long)mm.get("count")));

} 
}
注意：直接用 Jms 接口时接收了消息后要提交一下，否则下次启动接收者时还可以收到旧数据。有了 JmsTemplate 就不用自己提交 session.commit() 了。如果使用了 PooledConnectionFactory 要把 apache-activemq-5.3.0/lib/optional/activemq-pool-5.3.0.jar 加到 classpath

后半截  ：http://www.tech-q.cn/thread-5452-1-1.html