ActiveMQ 即时通讯服务 浅析

packagecom.hoo.mq.jms;

importjavax.jms.Connection;

importjavax.jms.ConnectionFactory;

importjavax.jms.DeliveryMode;

importjavax.jms.Destination;

importjavax.jms.MessageProducer;

importjavax.jms.Session;

importjavax.jms.TextMessage;

importorg.apache.activemq.ActiveMQConnection;

importorg.apache.activemq.ActiveMQConnectionFactory;

/**

*<b>function:</b>消息发送者

*@authorhoojo

*@createDate2013-6-19上午11:26:43

*@fileMessageSender.java

*@packagecom.hoo.mq.jms

*@projectActiveMQ-5.8

*@blog'target='_blank'>http://blog.csdn.net/IBM_hoojo[/code]
*@emailhoojo_@126.com
*@version1.0
*/
publicclassMessageSender{
//发送次数
publicstaticfinalintSEND_NUM=5;
//tcp地址
publicstaticfinalStringBROKER_URL="tcp://localhost:61616";
//目标，在ActiveMQ管理员控制台创建'target='_blank'>http://localhost:8161/admin/queues.jsp[/code]
publicstaticfinalStringDESTINATION="hoo.mq.queue";
/**
*<b>function:</b>发送消息
*@authorhoojo
*@createDate2013-6-19下午12:05:42
*@paramsession
*@paramproducer
*@throwsException
*/
publicstaticvoidsendMessage(Sessionsession,MessageProducerproducer)throwsException{
for(inti=0;i<SEND_NUM;i++){
Stringmessage="发送消息第"+(i+1)+"条";
TextMessagetext=session.createTextMessage(message);
System.out.println(message);
producer.send(text);
}
}
publicstaticvoidrun()throwsException{
Connectionconnection=null;
Sessionsession=null;
try{
//创建链接工厂
ConnectionFactoryfactory=newActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,ActiveMQConnection.DEFAULT_PASSWORD,BROKER_URL);
//通过工厂创建一个连接
connection=factory.createConnection();
//启动连接
connection.start();
//创建一个session会话
session=connection.createSession(Boolean.TRUE,Session.AUTO_ACKNOWLEDGE);
//创建一个消息队列
Destinationdestination=session.createQueue(DESTINATION);
//创建消息***者
MessageProducerproducer=session.createProducer(destination);
//设置持久化模式
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
sendMessage(session,producer);
//提交会话
session.commit();
}catch(Exceptione){
throwe;
}finally{
//关闭释放资源
if(session!=null){
session.close();
}
if(connection!=null){
connection.close();
}
}
}
publicstaticvoidmain(String[]args)throwsException{
MessageSender.run();
}
}


接受者
packagecom.hoo.mq.jms;
importjavax.jms.Connection;
importjavax.jms.ConnectionFactory;
importjavax.jms.Destination;
importjavax.jms.Message;
importjavax.jms.MessageConsumer;
importjavax.jms.Session;
importjavax.jms.TextMessage;
importorg.apache.activemq.ActiveMQConnection;
importorg.apache.activemq.ActiveMQConnectionFactory;
/**
*<b>function:</b>消息接收者
*@authorhoojo
*@createDate2013-6-19下午01:34:27
*@fileMessageReceiver.java
*@packagecom.hoo.mq.jms
*@projectActiveMQ-5.8
*@blog'target='_blank'>http://blog.csdn.net/IBM_hoojo[/code]
*@emailhoojo_@126.com
*@version1.0
*/
publicclassMessageReceiver{
//tcp地址
publicstaticfinalStringBROKER_URL="tcp://localhost:61616";
//目标，在ActiveMQ管理员控制台创建'target='_blank'>http://localhost:8161/admin/queues.jsp[/code]
publicstaticfinalStringDESTINATION="hoo.mq.queue";
publicstaticvoidrun()throwsException{
Connectionconnection=null;
Sessionsession=null;
try{
//创建链接工厂
ConnectionFactoryfactory=newActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,ActiveMQConnection.DEFAULT_PASSWORD,BROKER_URL);
//通过工厂创建一个连接
connection=factory.createConnection();
//启动连接
connection.start();
//创建一个session会话
session=connection.createSession(Boolean.TRUE,Session.AUTO_ACKNOWLEDGE);
//创建一个消息队列
Destinationdestination=session.createQueue(DESTINATION);
//创建消息***者
MessageConsumerconsumer=session.createConsumer(destination);
while(true){
//接收数据的时间（等待）100ms
Messagemessage=consumer.receive(1000*100);
TextMessagetext=(TextMessage)message;
if(text!=null){
System.out.println("接收："+text.getText());
}else{
break;
}
}
//提交会话
session.commit();
}catch(Exceptione){
throwe;
}finally{
//关闭释放资源
if(session!=null){
session.close();
}
if(connection!=null){
connection.close();
}
}
}
publicstaticvoidmain(String[]args)throwsException{
MessageReceiver.run();
}
}


2、Queue队列方式发送点对点消息数据
发送方
packagecom.hoo.mq.queue;
importjavax.jms.DeliveryMode;
importjavax.jms.MapMessage;
importjavax.jms.Queue;
importjavax.jms.QueueConnection;
importjavax.jms.QueueConnectionFactory;
importjavax.jms.QueueSession;
importjavax.jms.Session;
importorg.apache.activemq.ActiveMQConnection;
importorg.apache.activemq.ActiveMQConnectionFactory;
/**
*<b>function:</b>Queue方式消息发送者
*@authorhoojo
*@createDate2013-6-19下午04:34:36
*@fileQueueSender.java
*@packagecom.hoo.mq.queue
*@projectActiveMQ-5.8
*@blog'target='_blank'>http://blog.csdn.net/IBM_hoojo[/code]
*@emailhoojo_@126.com
*@version1.0
*/
publicclassQueueSender{
//发送次数
publicstaticfinalintSEND_NUM=5;
//tcp地址
publicstaticfinalStringBROKER_URL="tcp://localhost:61616";
//目标，在ActiveMQ管理员控制台创建'target='_blank'>http://localhost:8161/admin/queues.jsp[/code]
publicstaticfinalStringDESTINATION="hoo.mq.queue";
/**
*<b>function:</b>发送消息
*@authorhoojo
*@createDate2013-6-19下午12:05:42
*@paramsession
*@paramsender
*@throwsException
*/
publicstaticvoidsendMessage(QueueSessionsession,javax.jms.QueueSendersender)throwsException{
for(inti=0;i<SEND_NUM;i++){
Stringmessage="发送消息第"+(i+1)+"条";
MapMessagemap=session.createMapMessage();
map.setString("text",message);
map.setLong("time",System.currentTimeMillis());
System.out.println(map);
sender.send(map);
}
}
publicstaticvoidrun()throwsException{
QueueConnectionconnection=null;
QueueSessionsession=null;
try{
//创建链接工厂
QueueConnectionFactoryfactory=newActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,ActiveMQConnection.DEFAULT_PASSWORD,BROKER_URL);
//通过工厂创建一个连接
connection=factory.createQueueConnection();
//启动连接
connection.start();
//创建一个session会话
session=connection.createQueueSession(Boolean.TRUE,Session.AUTO_ACKNOWLEDGE);
//创建一个消息队列
Queuequeue=session.createQueue(DESTINATION);
//创建消息发送者
javax.jms.QueueSendersender=session.createSender(queue);
//设置持久化模式
sender.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
sendMessage(session,sender);
//提交会话
session.commit();
}catch(Exceptione){
throwe;
}finally{
//关闭释放资源
if(session!=null){
session.close();
}
if(connection!=null){
connection.close();
}
}
}
publicstaticvoidmain(String[]args)throwsException{
QueueSender.run();
}
}

接收方
packagecom.hoo.mq.queue;
importjavax.jms.JMSException;
importjavax.jms.MapMessage;
importjavax.jms.Message;
importjavax.jms.MessageListener;
importjavax.jms.Queue;
importjavax.jms.QueueConnection;
importjavax.jms.QueueConnectionFactory;
importjavax.jms.QueueSession;
importjavax.jms.Session;
importorg.apache.activemq.ActiveMQConnection;
importorg.apache.activemq.ActiveMQConnectionFactory;
/**
*<b>function:</b>消息接收者；依赖hawtbuf-1.9.jar
*@authorhoojo
*@createDate2013-6-19下午01:34:27
*@fileMessageReceiver.java
*@packagecom.hoo.mq.queue
*@projectActiveMQ-5.8
*@blog'target='_blank'>http://blog.csdn.net/IBM_hoojo[/code]
*@emailhoojo_@126.com
*@version1.0
*/
publicclassQueueReceiver{
//tcp地址
publicstaticfinalStringBROKER_URL="tcp://localhost:61616";
//目标，在ActiveMQ管理员控制台创建'target='_blank'>http://localhost:8161/admin/queues.jsp[/code]
publicstaticfinalStringTARGET="hoo.mq.queue";
publicstaticvoidrun()throwsException{
QueueConnectionconnection=null;
QueueSessionsession=null;
try{
//创建链接工厂
QueueConnectionFactoryfactory=newActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,ActiveMQConnection.DEFAULT_PASSWORD,BROKER_URL);
//通过工厂创建一个连接
connection=factory.createQueueConnection();
//启动连接
connection.start();
//创建一个session会话
session=connection.createQueueSession(Boolean.TRUE,Session.AUTO_ACKNOWLEDGE);
//创建一个消息队列
Queuequeue=session.createQueue(TARGET);
//创建消息***者
javax.jms.QueueReceiverreceiver=session.createReceiver(queue);
receiver.setMessageListener(newMessageListener(){
publicvoidonMessage(Messagemsg){
if(msg!=null){
MapMessagemap=(MapMessage)msg;
try{
System.out.println(map.getLong("time")+"接收#"+map.getString("text"));
}catch(JMSExceptione){
e.printStackTrace();
}
}
}
});
//休眠100ms再关闭
Thread.sleep(1000*100);
//提交会话
session.commit();
}catch(Exceptione){
throwe;
}finally{
//关闭释放资源
if(session!=null){
session.close();
}
if(connection!=null){
connection.close();
}
}
}
publicstaticvoidmain(String[]args)throwsException{
QueueReceiver.run();
}
}


3、Topic主题发布和订阅消息
消息发送方
packagecom.hoo.mq.topic;
importjavax.jms.DeliveryMode;
importjavax.jms.MapMessage;
importjavax.jms.Session;
importjavax.jms.Topic;
importjavax.jms.TopicConnection;
importjavax.jms.TopicConnectionFactory;
importjavax.jms.TopicPublisher;
importjavax.jms.TopicSession;
importorg.apache.activemq.ActiveMQConnection;
importorg.apache.activemq.ActiveMQConnectionFactory;
/**
*<b>function:</b>Queue方式消息发送者
*@authorhoojo
*@createDate2013-6-19下午04:34:36
*@fileQueueSender.java
*@packagecom.hoo.mq.topic
*@projectActiveMQ-5.8
*@blog'target='_blank'>http://blog.csdn.net/IBM_hoojo[/code]
*@emailhoojo_@126.com
*@version1.0
*/
publicclassTopicSender{
//发送次数
publicstaticfinalintSEND_NUM=5;
//tcp地址
publicstaticfinalStringBROKER_URL="tcp://localhost:61616";
//目标，在ActiveMQ管理员控制台创建'target='_blank'>http://localhost:8161/admin/queues.jsp[/code]
publicstaticfinalStringDESTINATION="hoo.mq.topic";
/**
*<b>function:</b>发送消息
*@authorhoojo
*@createDate2013-6-19下午12:05:42
*@paramsession会话
*@parampublisher发布者
*@throwsException
*/
publicstaticvoidsendMessage(TopicSessionsession,TopicPublisherpublisher)throwsException{
for(inti=0;i<SEND_NUM;i++){
Stringmessage="发送消息第"+(i+1)+"条";
MapMessagemap=session.createMapMessage();
map.setString("text",message);
map.setLong("time",System.currentTimeMillis());
System.out.println(map);
publisher.send(map);
}
}
publicstaticvoidrun()throwsException{
TopicConnectionconnection=null;
TopicSessionsession=null;
try{
//创建链接工厂
TopicConnectionFactoryfactory=newActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,ActiveMQConnection.DEFAULT_PASSWORD,BROKER_URL);
//通过工厂创建一个连接
connection=factory.createTopicConnection();
//启动连接
connection.start();
//创建一个session会话
session=connection.createTopicSession(Boolean.TRUE,Session.AUTO_ACKNOWLEDGE);
//创建一个消息队列
Topictopic=session.createTopic(DESTINATION);
//创建消息发送者
TopicPublisherpublisher=session.createPublisher(topic);
//设置持久化模式
publisher.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
sendMessage(session,publisher);
//提交会话
session.commit();
}catch(Exceptione){
throwe;
}finally{
//关闭释放资源
if(session!=null){
session.close();
}
if(connection!=null){
connection.close();
}
}
}
publicstaticvoidmain(String[]args)throwsException{
TopicSender.run();
}
}

接收方
packagecom.hoo.mq.topic;
importjavax.jms.JMSException;
importjavax.jms.MapMessage;
importjavax.jms.Message;
importjavax.jms.MessageListener;
importjavax.jms.Session;
importjavax.jms.Topic;
importjavax.jms.TopicConnection;
importjavax.jms.TopicConnectionFactory;
importjavax.jms.TopicSession;
importjavax.jms.TopicSubscriber;
importorg.apache.activemq.ActiveMQConnection;
importorg.apache.activemq.ActiveMQConnectionFactory;
/**
*<b>function:</b>消息接收者；依赖hawtbuf-1.9.jar
*@authorhoojo
*@createDate2013-6-19下午01:34:27
*@fileMessageReceiver.java
*@packagecom.hoo.mq.topic
*@projectActiveMQ-5.8
*@blog'target='_blank'>http://blog.csdn.net/IBM_hoojo[/code]
*@emailhoojo_@126.com
*@version1.0
*/
publicclassTopicReceiver{
//tcp地址
publicstaticfinalStringBROKER_URL="tcp://localhost:61616";
//目标，在ActiveMQ管理员控制台创建'target='_blank'>http://localhost:8161/admin/queues.jsp[/code]
publicstaticfinalStringTARGET="hoo.mq.topic";
publicstaticvoidrun()throwsException{
TopicConnectionconnection=null;
TopicSessionsession=null;
try{
//创建链接工厂
TopicConnectionFactoryfactory=newActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,ActiveMQConnection.DEFAULT_PASSWORD,BROKER_URL);
//通过工厂创建一个连接
connection=factory.createTopicConnection();
//启动连接
connection.start();
//创建一个session会话
session=connection.createTopicSession(Boolean.TRUE,Session.AUTO_ACKNOWLEDGE);
//创建一个消息队列
Topictopic=session.createTopic(TARGET);
//创建消息***者
TopicSubscribersubscriber=session.createSubscriber(topic);
subscriber.setMessageListener(newMessageListener(){
publicvoidonMessage(Messagemsg){
if(msg!=null){
MapMessagemap=(MapMessage)msg;
try{
System.out.println(map.getLong("time")+"接收#"+map.getString("text"));
}catch(JMSExceptione){
e.printStackTrace();
}
}
}
});
//休眠100ms再关闭
Thread.sleep(1000*100);
//提交会话
session.commit();
}catch(Exceptione){
throwe;
}finally{
//关闭释放资源
if(session!=null){
session.close();
}
if(connection!=null){
connection.close();
}
}
}
publicstaticvoidmain(String[]args)throwsException{
TopicReceiver.run();
}
}


4、整合Spring实现消息发送和接收，在整合之前我们需要添加jar包，需要的jar包如下



这些jar包可以在D:\apache-activemq-5.8.0\lib这个lib目录中找到，添加完jar包后就开始编码工作。
消息发送者
packagecom.hoo.mq.spring.support;
importjava.util.Date;
importjavax.jms.JMSException;
importjavax.jms.MapMessage;
importjavax.jms.Message;
importjavax.jms.Session;
importorg.springframework.context.ApplicationContext;
importorg.springframework.context.support.FileSystemXmlApplicationContext;
importorg.springframework.jms.core.JmsTemplate;
importorg.springframework.jms.core.MessageCreator;
/**
*<b>function:</b>SpringJMSTemplate消息发送者
*@authorhoojo
*@createDate2013-6-24下午02:18:48
*@fileSender.java
*@packagecom.hoo.mq.spring.support
*@projectActiveMQ-5.8
*@blog'target='_blank'>http://blog.csdn.net/IBM_hoojo[/code]
*@emailhoojo_@126.com
*@version1.0
*/
publicclassSender{
publicstaticvoidmain(String[]args){
ApplicationContextctx=newFileSystemXmlApplicationContext("classpath:applicationContext-*.xml");
JmsTemplatejmsTemplate=(JmsTemplate)ctx.getBean("jmsTemplate");
jmsTemplate.send(newMessageCreator(){
publicMessagecreateMessage(Sessionsession)throwsJMSException{
MapMessagemessage=session.createMapMessage();
message.setString("message","currentsystemtime:"+newDate().getTime());
returnmessage;
}
});
}
}

消息接收者
packagecom.hoo.mq.spring.support;
importjava.util.Map;
importorg.springframework.context.ApplicationContext;
importorg.springframework.context.support.FileSystemXmlApplicationContext;
importorg.springframework.jms.core.JmsTemplate;
/**
*<b>function:</b>SpringJMSTemplate消息接收者
*@authorhoojo
*@createDate2013-6-24下午02:22:32
*@fileReceiver.java
*@packagecom.hoo.mq.spring.support
*@projectActiveMQ-5.8
*@blog'target='_blank'>http://blog.csdn.net/IBM_hoojo[/code]
*@emailhoojo_@126.com
*@version1.0
*/
publicclassReceiver{
@SuppressWarnings("unchecked")
publicstaticvoidmain(String[]args){
ApplicationContextctx=newFileSystemXmlApplicationContext("classpath:applicationContext-*.xml");
JmsTemplatejmsTemplate=(JmsTemplate)ctx.getBean("jmsTemplate");
while(true){
Map<String,Object>map=(Map<String,Object>)jmsTemplate.receiveAndConvert();
System.out.println("收到消息："+map.get("message"));
}
}
}

这里主要是用到了JmsTemplate这个消息模板，这个对象在spring的IoC容器中管理，所以要从spring的容器上下文中获取。下面看看spring的配置文件applicationContext-beans.xml内容：
<?xmlversion="1.0"encoding="UTF-8"?>
<beansxmlns="http://www.springframework.org/schema/beans"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans
'target='_blank'>http://www.springframework.org/schema/beans/spring-beans-3.1.xsd[/code]
'target='_blank'>http://www.springframework.org/schema/context[/code]
'target='_blank'>http://www.springframework.org/schema/context/spring-context-3.1.xsd">[/code]
<!--连接池-->
<beanid="pooledConnectionFactory"class="org.apache.activemq.pool.PooledConnectionFactory"destroy-method="stop">
<propertyname="connectionFactory">
<beanclass="org.apache.activemq.ActiveMQConnectionFactory">
<propertyname="brokerURL"value="tcp://localhost:61616"/>
</bean>
</property>
</bean>
<!--连接工厂-->
<beanid="activeMQConnectionFactory"class="org.apache.activemq.ActiveMQConnectionFactory">
<propertyname="brokerURL"value="tcp://localhost:61616"/>
</bean>
<!--配置消息目标-->
<beanid="destination"class="org.apache.activemq.command.ActiveMQQueue">
<!--目标，在ActiveMQ管理员控制台创建http://localhost:8161/admin/queues.jsp-->
<constructor-argindex="0"value="hoo.mq.queue"/>
</bean>
<!--消息模板-->
<beanid="jmsTemplate"class="org.springframework.jms.core.JmsTemplate">
<propertyname="connectionFactory"ref="activeMQConnectionFactory"/>
<propertyname="defaultDestination"ref="destination"/>
<propertyname="messageConverter">
<beanclass="org.springframework.jms.support.converter.SimpleMessageConverter"/>
</property>
</bean>
</beans>

这里的整合就比较简单了，如果你是web工程，那你在需要用jms的时候，只需用注入jmsTemplate即可。