spring-amqp整合rabbitmq消费者配置和代码

标签：

<?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:context="http://www.springframework.org/schema/context"
       xmlns:rabbit="http://www.springframework.org/schema/rabbit"
       xsi:schemaLocation="
            http://www.springframework.org/schema/beans
                http://www.springframework.org/schema/beans/spring-beans.xsd
            http://www.springframework.org/schema/context
                http://www.springframework.org/schema/context/spring-context.xsd
            http://www.springframework.org/schema/rabbit
                http://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd">
 
     
   <!-- 连接服务配置  -->
   <rabbit:connection-factory id="mqConnectionFactory" host="localhost" username="guest"
        password="guest" port="5672"  />
         
   <rabbit:admin connection-factory="mqConnectionFactory"/>
    
   <!-- queue 队列声明-->
   <rabbit:queue id="queue_one" durable="true" auto-delete="false" exclusive="false" name="queue_one"/>
    
    <!-- 配置线程池 -->  
<bean id ="taskExecutor"  class ="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor" >  
    <!-- 线程池维护线程的最少数量 -->  
<property name ="corePoolSize" value ="5" />  
    <!-- 线程池维护线程所允许的空闲时间 -->  
<property name ="keepAliveSeconds" value ="30000" />  
    <!-- 线程池维护线程的最大数量 -->  
<property name ="maxPoolSize" value ="1000" />  
    <!-- 线程池所使用的缓冲队列 -->  
<property name ="queueCapacity" value ="200" />  
</bean>  
   <!-- exchange queue binging key 绑定 -->
    <rabbit:direct-exchange name="my-mq-exchange" durable="true" auto-delete="false" id="my-mq-exchange">
        <rabbit:bindings>
            <rabbit:binding queue="queue_one" key="queue_one_key"/>
        </rabbit:bindings>
    </rabbit:direct-exchange>
       <!-- 消息监听器 -->  
    <bean id="consumerMessageListener" class="com.netease.mobileMq.task.deviceCacheFlushTask"/>  
    <!-- 可以获取session的MessageListener -->  
    <!-- queue litener  观察 监听模式 当有消息到达时会通知监听在对应的队列上的监听对象-->
    <rabbit:listener-container connection-factory="mqConnectionFactory" acknowledge="auto" task-executor="taskExecutor">
        <rabbit:listener queues="queue_one" ref="consumerMessageListener"/>
    </rabbit:listener-container>
</beans>

package com.netease.mobileMq.task;

import java.util.Date;
import java.util.List;

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsTemplate;

import com.alibaba.fastjson.JSON;
import com.netease.commonBean.FlushDeviceCacheBean;
import com.netease.device.constant.EquipmentConstants;
import com.netease.device.dao.EquipmentMapper;
import com.netease.device.dao.FingerUserMapper;
import com.netease.device.entity.EquipmentInfo;
import com.netease.device.entity.FingerUserInfo;
import com.netease.mobile.common.RedisUtil;

/**
 * @author 作者 E-mail:ruanjianlxm@sina.com
 * @version 创建时间：2015年8月4日 下午4:44:39
 * 类说明
 */
public class deviceCacheFlushTask implements  
MessageListener{
	private static Logger logger = LoggerFactory.getLogger("equipmentErrorLog");
	@Autowired
	FingerUserMapper fingerUserMapper;
	@Autowired
	EquipmentMapper equipmentMapper;
	@Override
	public void onMessage(Message  message){
		 
		 String receiveMsg=null;
		try {
			receiveMsg =new String(message.getBody(),"utf-8");
		} catch (Exception e1) {
			// TODO Auto-generated catch block
			e1.printStackTrace();
			return ;
		}
		System.out.println("receiveMsg:"+receiveMsg);
		if (StringUtils.isBlank(receiveMsg)) {
			logger.error("deviceCacheFlushTask   receiveMsg is null  Time is " + new Date());
			return ;
		}
		else {
			logger.info("deviceCacheFlushTask   receiveMsg " + receiveMsg);//日志中记录每个刷新的数据
			
		}
		
		FlushDeviceCacheBean flushBean = JSON.parseObject(receiveMsg, FlushDeviceCacheBean.class);//将传过来的刷新对象进行格式化。
		String mainssn = flushBean.getMainssn();
		String[] refIds = flushBean.getUserIds();
		
		/*---------------先更新用户缓存----------*/
		if (mainssn != null) {//有主账号就更新主账号信息
		List<FingerUserInfo> fingerUserInfos = null;
		try {
			fingerUserInfos = fingerUserMapper.getAllEqUserInfoByName(mainssn);
		} catch (Exception e) {
			// TODO: handle exception
			logger.error("EquipmentServiceImpl flushCache error", e);
			return ;
		}

		if (fingerUserInfos == null || fingerUserInfos.size() == 0) {// 如果有一个都没有
			RedisUtil.delete(EquipmentConstants.EQUIPMENT_FINGER_USER_PRE
					+ mainssn);
		} else {
			RedisUtil.set(EquipmentConstants.EQUIPMENT_FINGER_USER_PRE
					+ mainssn, JSON.toJSONString(fingerUserInfos));// 新数据直接替换到缓存中
		}
		}
		/*---------------再更新设备缓存----------*/

		List<EquipmentInfo> equipmentInfos = null;
		
		if (refIds!=null&&refIds.length!=0) {//有ID就刷新ID信息
		for (String refId : refIds) {
			try {
				equipmentInfos = equipmentMapper.getAllEquipmentInfoById(refId);
			} catch (Exception e) {
				// TODO: handle exception
				logger.error("EquipmentServiceImpl flushCache error", e);
				return ;
			}
			if (equipmentInfos == null || equipmentInfos.size() == 0) {// 如果有一个没有，说明查询数据出错失败了
				RedisUtil.delete(EquipmentConstants.EQUIPMENT_EQUINFO_PRE + refId);
			} else {
				RedisUtil.set(EquipmentConstants.EQUIPMENT_EQUINFO_PRE + refId,
						JSON.toJSONString(equipmentInfos));// 新数据直接替换到缓存中
			}
		}
		}
	}

}

spring-amqp整合rabbitmq消费者配置和代码

标签：

原文地址：http://my.oschina.net/91jason/blog/488126