Spring + ActiveMQ 实现

站内消息模块在多数OA系统中是不可缺少的一部分。

     一般来说，是消息是存储在数据库中的，但最关键的是消息发送及接收处理的环节。如果直接用一个Action方法处理，然后调用Service来存储消息，那样也是可以的，问题是等待的时间较长，而且耗费服务器的资源。这种方法相当于同步处理。

     因此，采用另一外种方法：异步处理。说到这里，想到twitter的系统，虽然它是采用Ruby On Rails + Scala等实现的，但思想是相同的！客户端发送一个信息（任务）过来，后台把任务加入队列，队列是由若干个守护进程处理的！这样不仅能实时处理客户的其 它请求，也能充分利用系统的资源。

     那么下面开始简要写一下Spring中的JMS配置。注意，运行时要启动ActiveMQ的服务器~~，，当然你也可以自己配置嵌入服务器的ActiveMQ.。

     添加一个applicationContext-jms.xml文件。

 <!-- ActiveMQ 连接 -->
    <bean id="connectionFactory"
            class="org.apache.activemq.ActiveMQConnectionFactory">
        <property name="brokerURL" value="tcp://localhost:61616" />
        <property name="useAsyncSend" value="true"/>
    </bean>
    <bean id="destination" 
            class="org.apache.activemq.command.ActiveMQQueue">
            <constructor-arg index="0" value="myqueue"></constructor-arg>
    </bean>
    
    <bean id="jmsTemplate"
            class="org.springframework.jms.core.JmsTemplate">
        <property name="connectionFactory" ref="connectionFactory"></property>
        <property name="defaultDestination" ref="destination"></property>
        <property name="messageConverter" ref="innerMessageConverter"></property>
    </bean>
    
    <bean id="innerMessageConverter" class="jmu.xmpg.service.jms.InnerMessageConverter"></bean>
    
    <bean id="messageService" class="jmu.xmpg.service.jms.MessageServiceImpl">
        <!-- <property name="jmsTemplate" ref="jmsTemplate"></property> -->
    </bean>
    
    <bean id="messageListener" class="jmu.xmpg.service.jms.MessageMDB"></bean>
     
    <bean id="jmsContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer" lazy-init="false">
        <property name="connectionFactory" ref="connectionFactory"></property>
        <property name="destination" ref="destination"></property>
        <property name="messageListener" ref="messageListener"></property>
        <property name="concurrentConsumers" value="5"></property>
        <!--0:CACHE_NONE,1:CACHE_CONNECTION,2:CACHE_SESSION,3:CACHE_CONSUMER,4:CACHE_AUTO-->
        <property name="cacheLevel" value="0"/>
    </bean>

下面简单说明一下各个bean：
connectionFactory：用来连接ActiveMQ的连接Bean
destination：用来作为信息发送的目的地队列。
jmsTemplate: Spring封装的JMS发送接收模板。
messageConverter: 对接收和发送的消息进行转换，写一次，就能省下每次的自己手动转换，:D
messageService：消息Service，用来调用发送消息。
messageListener: 消息驱动Bean，在EJB中称为MDB，用来监听队列中的消息，异步接收处理消息
jmsContainer：消息监听容器，当消息到达时，将消息转给messageListener，而且可以设置多个消费者。想起操作系统中的 消费者-生产者的问题了，，哈。
      这里，列出主要使用的代码：
转换器InnerMessageConverter.java

public class InnerMessageConverter implements MessageConverter {
public InnerMessageConverter(){}

@Override
public Object fromMessage(Message message) throws JMSException,
MessageConversionException {
  if(!(message instanceof MapMessage)){
    throw new MessageConversionException("Message isn't a MapMessage");
  }
  MapMessage msg = (MapMessage)message;
  MessageDescription messageDescription = new MessageDescription();
  messageDescription.setContent(msg.getString("content"));
  messageDescription.setCreateDate(new Date(msg.getLong("createDate")));
  messageDescription.setLevel(msg.getShort("level"));
  messageDescription.setToUserIds(msg.getString("toUserIds"));
  messageDescription.setFromUserName(msg.getString("fromUserName"));
  return messageDescription;
}

@Override
public Message toMessage(Object object, Session session) throws JMSException,
  MessageConversionException {
  if(!(object instanceof MessageDescription)){
    throw new MessageConversionException("Object isn't a MessageDescription");
  }
  MessageDescription messageDescription = (MessageDescription)object;
  MapMessage message = session.createMapMessage();
  message.setString("content", messageDescription.getContent());
  message.setLong("createDate", messageDescription.getCreateDate().getTime());
  message.setShort("level", messageDescription.getLevel());
  message.setString("toUserIds", messageDescription.getToUserIds());
  message.setString("fromUserName", messageDescription.getFromUserName());
  return message;
 }
}

消息Service：MessageServiceImpl.java
MessageService接口为自定义，主要有recevie和send方法，这里就不列出。
注意看 Send方法，简单吧:D，有加入转换器，代码就是少~~

package jmu.xmpg.service.jms;
import jmu.xmpg.dao.jms.InnerMessageDao;
import jmu.xmpg.entity.jms.InnerMessage;
import jmu.xmpg.entity.jms.MessageDescription;
import jmu.xmpg.entity.user.User;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.transaction.annotation.Transactional;
import org.springside.modules.orm.Page;
/**
*
* @date : Jun 26, 2009 5:16:44 PM
* Filename ：MessageServiceImpl.java
* Author: SloanWu
*/
@Transactional(readOnly=true)
public class MessageServiceImpl implements MessageService{
@Autowired
private JmsTemplate jmsTemplate;

@Autowired
private InnerMessageDao innerMessageDao;

@Override
public Page<InnerMessage> recevie(User user, Page<InnerMessage> page){
  page = innerMessageDao.getAll(page);
  return page;
}

@Override
public void send(final MessageDescription mesinfo){
    jmsTemplate.convertAndSend(mesinfo);
  }
}

下面一个是最主要的代码：MessageMDB.java
主要方法为onMessage:用来转换消息，processMessage，用来处理转换后的消息，可以用Spring的消息代理来简化这个MDB，并且可以不用实现MessageListener接口，纯POJO！

package jmu.xmpg.service.jms;
import java.util.Date;
import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.MessageListener;
import jmu.xmpg.dao.jms.InnerMessageDao;
import jmu.xmpg.dao.jms.MessageDescriptionDao;
import jmu.xmpg.dao.user.UserDao;
import jmu.xmpg.entity.jms.InnerMessage;
import jmu.xmpg.entity.jms.MessageDescription;
import jmu.xmpg.entity.user.User;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.support.converter.MessageConversionException;
import org.springframework.transaction.annotation.Transactional;
import freemarker.template.utility.StringUtil;
/**
*
* @date : Jun 26, 2009 5:44:26 PM
* Filename ：MessageMDB.java
* Author: SloanWu
*/
//默认将类中的所有函数纳入事务管理.
@Transactional
public class MessageMDB implements MessageListener{

@Autowired
private InnerMessageDao innerMessageDao;

@Autowired
private MessageDescriptionDao messageDescriptionDao;

@Autowired
private UserDao userDao;

@Override
public void onMessage(Message message) {
try {
    if(!(message instanceof MapMessage)){
      throw new MessageConversionException("Message isn't a MapMessage");
    }
    MapMessage msg = (MapMessage)message;
    MessageDescription messageDescription = new MessageDescription();
    messageDescription.setContent(msg.getString("content"));
    messageDescription.setCreateDate(new Date(msg.getLong("createDate")));
    messageDescription.setLevel(msg.getShort("level"));
    messageDescription.setToUserIds(msg.getString("toUserIds"));
    messageDescription.setFromUserName(msg.getString("fromUserName"));
    processMessage(messageDescription);
  } catch (Exception e) {
    System.out.println("JMS Exception:"+e.toString());
  }
}

private void processMessage(MessageDescription mesDescription){
  testShow(mesDescription);
  User fromUser = userDao.loadByLoginName(mesDescription.getFromUserName());
  mesDescription.setFromUser(fromUser);
  messageDescriptionDao.save(mesDescription);
  String[] ids = StringUtil.split(mesDescription.getToUserIds(), ',');
  for (int i = 0; i < ids.length; i++) {
    User toUser = new User();
    toUser.setId(Long.valueOf(ids[i]));
    InnerMessage inner = new InnerMessage();
    inner.setMessageDescription(mesDescription);
    inner.setReadable(false);
    inner.setToUser(toUser);
    innerMessageDao.save(inner);
  }
}
  private void testShow(MessageDescription mesDescription) {
    System.out.println("--收到消息："+mesDescription.getFromUserName());
    System.out.println("-----时间："+mesDescription.getCreateDate());
    System.out.println("--消息等级："+mesDescription.getLevel());
    System.out.println("--消息内容："+mesDescription.getContent());
    System.out.println("--接收用户："+mesDescription.getToUserIds());
  }
}

评论

2 楼 guojiew 2015-07-07  

只是剪切了一部分代码出来，看的真是脑袋大。

1 楼 Qudong_0129 2013-03-13  

哥们能把代码贴全不? 或者上传个附件..只看这些看的我云里雾里的...谢谢了!