jms 学习笔记1

今天有学习一下JMS,从一个简单例子开始

名词概念: 消息中介ActiveMQ    消息   队列   主题

  JMS定义了Java中访问消息中间件的接口

  JMS只是接口，并没有给予实现，实现JMS接口的消息中间件叫JMS   Provider，这样的消息中间件可以从Java里通过JMS接口进行调用。         
  JMS结构：header和body。header包含消息的识别信息和路由信息，body包含消息的实际数据。         
   JMS消息分类：    
  BytesMessage   消息是字节流。    
  MapMessage   消息是一系列的命名和值的对应组合。    
  ObjectMessage   消息是一个流化的Java对象。    
  StreamMessage   消息是Java中的输入输出流。    
  TextMessage   消息是一个字符串，这种类型将会广泛用于XML格式的数据。

下面是一个点对点模式的JMS例子:

建立WEB工程 导入ActiveMQ Spring 相关jar包,工程相关文件如下:

web.xml

 

<?xml version="1.0" encoding="UTF-8"?>
<web-app xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance "
 xmlns="http://java.sun.com/xml/ns/javaee " xmlns:web="http://java.sun.com/xml/ns/javaee/web-app_2_5.xsd "
 xsi:schemaLocation="http://java.sun.com/xml/ns/javaee http://java.sun.com/xml/ns/javaee/web-app_2_5.xsd "
 id="WebApp_ID" version="2.5">
 <display-name>jms</display-name>

 <listener>
  <listener-class>org.springframework.web.context.ContextLoaderListener</listener-class>
 </listener>
</web-app>

 

spring 文件

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans "
  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance "
  xmlns:aop="http://www.springframework.org/schema/aop "
  xmlns:tx="http://www.springframework.org/schema/tx "
  xsi:schemaLocation="
   http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
   http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-2.5.xsd
   http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-2.5.xsd ">
 <!-- 定义消息中介的连接工厂 -->
 <bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
  <property name="brokerURL" value="tcp://localhost:61616"/>
 </bean>
 
 <!-- 定义一个队列,存放消息 -->
 <bean id="kesnQDestination" class="org.apache.activemq.command.ActiveMQQueue">
  <constructor-arg index="0" value="kesn.queue"/>
 </bean>
 
 <!-- 定义一个主题,用于发布消息 -->
 <bean id="kesnTDestination" class="org.apache.activemq.command.ActiveMQTopic">
  <constructor-arg index="0" value="kesn.topic"/>
 </bean>
 
 <!-- JMS模板 -->
 <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
  <property name="connectionFactory" ref="connectionFactory"/>
 </bean>
 
 <!-- 消息接收者 -->
 <bean id="messageReceiver" class="com.kesn.jms.service.impl.KesnJmsTemplateMessageReceiver">
  <property name="jmsTemplate" ref="jmsTemplate"/>
  <property name="destination" ref="kesnQDestination"/>
 </bean>
 
 <!-- 消息发送者 -->
 <bean id="messageSender" class="com.kesn.jms.service.impl.KesnJmsTemplateMessageSender">
  <property name="jmsTemplate" ref="jmsTemplate"/>
  <property name="destination" ref="kesnQDestination"/>
 </bean>
</beans>

 

一个简单的发送消息和接收消息的类

 

package com.kesn.jms.demo;

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

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.log4j.Logger;

/**
 * 消息的发送者
 * @author LEPING LI
 * @version 1.0.0
 */
public class KesnMessageSender {
 private static Logger log=Logger.getLogger(KesnMessageSender.class);
 public static void main(String[] args) {
  ConnectionFactory cf=new ActiveMQConnectionFactory("tcp://localhost:61616");
  Connection con=null;
  Session session=null;
  try {
   con=cf.createConnection();
   //创建会话
   session=con.createSession(false, Session.AUTO_ACKNOWLEDGE);
   //创建队列
   Destination dest=new ActiveMQQueue("kesn.queue");

 

   //创建主题可以实现发布--订阅模式
   //Destination dest=new ActiveMQTopic("kesn.queue");

   //session是的消息生产者和发送者的工厂
   MessageProducer producer=session.createProducer(dest); //创建消息生产者
   TextMessage message=session.createTextMessage();  //创建消息
   message.setText("老婆,我爱你!"); 
   producer.send(message);         //发送消息
   log.info("消息发送成功!");
  } catch (JMSException e) {
   e.printStackTrace();
  }finally{
   try {
    if(session!=null){
     session.close();
    }
    if(con!=null){
     con.close();
    }
   } catch (JMSException e) {
    e.printStackTrace();
   }
  }
  
 }
}

 

package com.kesn.jms.demo;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.log4j.Logger;

/**
 * 消息接收者
 * @author LEPING LI
 *
 */
public class KesnMessageReceiver {
 private static Logger log=Logger.getLogger(KesnMessageReceiver.class);
 public static void main(String[] args) {
  ConnectionFactory cf=new ActiveMQConnectionFactory("tcp://localhost:61616");
  Connection con=null;
  Session session=null;
  try {
   con=cf.createConnection();
   //创建会话
   session=con.createSession(false, Session.AUTO_ACKNOWLEDGE);
   //创建队列
   Destination dest=new ActiveMQQueue("kesn.queue");

   //创建主题可以实现发布--订阅模式
   //Destination dest=new ActiveMQTopic("kesn.queue");
   MessageConsumer consumer=session.createConsumer(dest); //创建消息消费者
   con.start();
   Message message=consumer.receive();      //接收消息
   TextMessage textMessage=(TextMessage)message;   
   log.info("收到消息:"+textMessage.getText());
  } catch (JMSException e) {
   e.printStackTrace();
  }finally{
   try {
    if(session!=null){
     session.close();
    }
    if(con!=null){
     con.close();
    }
   } catch (JMSException e) {
    e.printStackTrace();
   }
  }
  
 
 }

 

由于JMS上面中存在大量的代码冗余,可以采用Spring 提示的JmsTemplate 来简化操作

接口类省略

 

package com.kesn.jms.service.impl;

import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.Session;

import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;

import com.kesn.jms.entity.Motorist;
import com.kesn.jms.service.JmsTemplateSendService;
/**
 * 使用JmsTemplate 发送消息
 * @author LEPING LI
 *
 */
public class KesnJmsTemplateMessageSender implements JmsTemplateSendService{
 private JmsTemplate jmsTemplate;
 public void setJmsTemplate(JmsTemplate jmsTemplate) {
  this.jmsTemplate = jmsTemplate;
 }
 private Destination destination;
 public void setDestination(Destination destination) {
  this.destination = destination;
 }
 
 @Override
 public void sendMessage(final Motorist m) {
  jmsTemplate.send(destination, new MessageCreator(){
   @Override
   public Message createMessage(Session session) throws JMSException {
    MapMessage message=session.createMapMessage();
    message.setString("name",m.getName());
    message.setString("email",m.getEmail());
    message.setString("age", m.getAge());
    return message;
   }
  });
 }
}

}

 

 

package com.kesn.jms.service.impl;

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

import org.springframework.jms.JmsException;
import org.springframework.jms.core.JmsTemplate;

import com.kesn.jms.entity.Motorist;
import com.kesn.jms.service.JmsTemplateReceiveService;
/**
 * 使用JmsTemplate 发送消息
 * @author LEPING LI
 *
 */
public class KesnJmsTemplateMessageReceiver implements JmsTemplateReceiveService{
 
 private JmsTemplate jmsTemplate;
 public void setJmsTemplate(JmsTemplate jmsTemplate) {
  this.jmsTemplate = jmsTemplate;
 }
 private Destination destination;
 public void setDestination(Destination destination) {
  this.destination = destination;
 }

 @Override
 public Motorist receiveMessage() {
  try {
   MapMessage message=(MapMessage) jmsTemplate.receive(destination);
   Motorist m=new Motorist();
   m.setName(message.getString("name"));
   m.setEmail(message.getString("email"));
   m.setAge(message.getString("age"));
   return m;
  } catch (JmsException e) {
   e.printStackTrace();
  } catch (JMSException e) {
   e.printStackTrace();
  }
  return null;
 }
}

 

测试类,SE环境下测试

package com.kesn.jms.client;

import org.apache.log4j.Logger;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

import com.kesn.jms.entity.Motorist;
import com.kesn.jms.service.impl.KesnJmsTemplateMessageSender;

public class MessageSendClient {
 private static Logger log=Logger.getLogger(MessageSendClient.class);
 public static void main(String[] args) {
  ApplicationContext context=new ClassPathXmlApplicationContext("applicationContext.xml");
  KesnJmsTemplateMessageSender sender=(KesnJmsTemplateMessageSender) context.getBean("messageSender");
  Motorist m=new Motorist();
  m.setName("李乐平");
  m.setEmail("lilpjob");
  m.setAge("1");
  sender.sendMessage(m);
  log.info("消息发送成功!");
 }
}

package com.kesn.jms.client;

import org.apache.log4j.Logger;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

import com.kesn.jms.entity.Motorist;
import com.kesn.jms.service.impl.KesnJmsTemplateMessageReceiver;

public class MessageReciveClient {
 private static Logger log=Logger.getLogger(MessageReciveClient.class);
 public static void main(String[] args) {
  ApplicationContext context=new ClassPathXmlApplicationContext("applicationContext.xml");
  KesnJmsTemplateMessageReceiver receiver=(KesnJmsTemplateMessageReceiver) context.getBean("messageReceiver");
  Motorist m=receiver.receiveMessage();
  log.info("name:"+m.getName());
  log.info("email:"+m.getEmail());
  log.info("age:"+m.getAge());
 }
}

 

 

 初学者问题颇多,在使用连接创建会话时,第一个参数boolean transacted  是否加以事务管理? 如果改为true ,客户端接收不到消息!还没有搞明白! 

 JMS称消息的异步传输,只是对于点对点的模式面言吗? 发布--订阅模式好像消息消费者只有在监听情况下才能接收到发布到主题的消息! ??