MQ消息接收接口

<beans xmlns="http://www.springframework.org/schema/beans"
	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:p="http://www.springframework.org/schema/p"
	xmlns:jee="http://www.springframework.org/schema/jee" xmlns:tx="http://www.springframework.org/schema/tx"
	xmlns:aop="http://www.springframework.org/schema/aop" xmlns:context="http://www.springframework.org/schema/context"
	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd   
       http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-3.0.xsd   
       http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-3.0.xsd   
       http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.0.xsd
       http://www.springframework.org/schema/jee http://www.springframework.org/schema/jee/spring-jee-2.0.xsd"
	default-lazy-init="false">
	
	<!-- MQ消息监听接口服务 -->
	<bean id="jmsReceiverListenerServer" class="com.iteye.jms.receive.JmsReceiverListenerServer"></bean>
	
	<!-- MQ队列连接工厂配置 -->
	<bean id="connectionFactory" class="com.ibm.mq.jms.MQQueueConnectionFactory"> 
		<property name="queueManager" value="队列管理器名称" />
		<property name="hostName" value="IP" /> 
		<property name="port" value="端口" /> 
		<property name="channel" value="通道" />
		<property name="transportType" value="传输方式" />
	</bean> 
	
	<!-- MQ队列连接工厂配置适配器 -->
	<bean id="connectionFactoryAdapter"
		class="org.springframework.jms.connection.UserCredentialsConnectionFactoryAdapter">
		<property name="targetConnectionFactory" ref="connectionFactory" />
	</bean>
	
	<!-- 订单接收队列 -->
	<bean id="orderReceiveQueue" class="com.ibm.mq.jms.MQQueue"> 
		<property name="baseQueueName" value="OrderReceive_Queue" /> 
	</bean> 
	
	<!-- 商品接收队列 -->
	<bean id="productReceiveQueue" class="com.ibm.mq.jms.MQQueue"> 
		<property name="baseQueueName" value="ProductReceive_Queue" /> 
	</bean> 
	
	<!-- 订单消息监听器 -->	
	<bean  class="com.iteye.jms.receive.OrderQueueMessageListener"></bean>
	
	<!-- 商品消息监听器 -->	
	<bean  class="com.iteye.jms.receive.ProductQueueMessageListener"></bean>
</beans>

package com.iteye.jms.receive;

import static java.lang.annotation.ElementType.TYPE;
import static java.lang.annotation.RetentionPolicy.RUNTIME;

import java.lang.annotation.Retention;
import java.lang.annotation.Target;

@Target(TYPE)
@Retention(RUNTIME)
public @interface JmsReceiveMsgAnno {

    /**
     * 监听的链接工厂
     */
    String connectionFactory() default "";

    /**
     * 监听队列
     */
    String receiveQueue() default "";

    /**
     * 监听端消费者数量
     */
    int consumers() default 1;

    /**
     * 消息选择器
     */
    String messageSelector() default "";

    /**
     * 是否开启事务
     */
    boolean sessionTransacted() default false;
}

package com.iteye.jms.receive;

import java.util.HashSet;
import java.util.Set;

import javax.jms.ConnectionFactory;
import javax.jms.MessageListener;
import javax.jms.Queue;

import org.apache.commons.lang.StringUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.jms.listener.DefaultMessageListenerContainer;

/**
 * 
 * 〈一句话功能简述〉<br>
 * MQ消息监听接口服务
 * 
 * @see [相关类/方法]（可选）
 * @since [产品/模块版本] （可选）
 */
@SuppressWarnings("rawtypes")
public class JmsReceiverListenerServer implements BeanPostProcessor, ApplicationContextAware, ApplicationListener,
        DisposableBean {

    private ApplicationContext applicationContext;

    private boolean started = false;

    /**
     * 服务端监听器容器列表(Set集合不允许重复)
     */
    private Set<DefaultMessageListenerContainer> messageListenerContainers = new HashSet<DefaultMessageListenerContainer>();

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    @Override
    public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
        // 扫描所有使用了@JmsReceiveMsgAnno的bean,将其注册为服务端监听器容器列表
        registerMQServerBean(bean);
        return bean;
    }

    @Override
    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
        return bean;
    }

    /**
     * 在所有的Spring加载工作完成之后会执行onApplicationEvent方法,启动消息监听器进行消息监听
     */
    @Override
    public void onApplicationEvent(ApplicationEvent event) {
        if (isCurrentApplicationContextRefresh(event)) {
            if (started) {
                return;
            }

            started = true;
            // 遍历所有消息监听器列表,启动消息监听
            for (DefaultMessageListenerContainer container : messageListenerContainers) {
                container.afterPropertiesSet();
                container.start();
            }
        }
    }

    /**
     * 
     * 是否是当前上下文，防止重复加载和过早加载 <br>
     * 〈功能详细描述〉
     * 
     * @param event
     * @return
     * @see [相关类/方法](可选)
     * @since [产品/模块版本](可选)
     */
    private boolean isCurrentApplicationContextRefresh(ApplicationEvent event) {
        return event instanceof ContextRefreshedEvent
                && ((ContextRefreshedEvent) event).getApplicationContext() == applicationContext;
    }

    /**
     * 容器销毁时停止消息监听器监听
     */
    @Override
    public void destroy() throws Exception {
        if (started) {
            for (DefaultMessageListenerContainer container : messageListenerContainers) {
                container.shutdown();
            }
        }
    }

    /**
     * 
     * 扫描所有使用了@JmsReceiveMsgAnno的bean,将其注册为服务端监听器容器列表 <br>
     * 〈功能详细描述〉
     * 
     * @param bean
     * @see [相关类/方法](可选)
     * @since [产品/模块版本](可选)
     */
    private void registerMQServerBean(Object bean) {
        Class<?> clazz = bean.getClass();
        // 该类是否使用了JmsReceiveMsgAnno注解
        if (clazz.isAnnotationPresent(JmsReceiveMsgAnno.class) == false) {
            return;
        } else {
            // 该类必须实现MessageListener接口
            if (!MessageListener.class.isAssignableFrom(clazz)) {
                throw new RuntimeException("消息监听器必须实现MessageListener接口");
            }
        }

        // 消息监听器
        MessageListener listener = (MessageListener) bean;
        // 注解信息类对象
        JmsReceiveMsgAnno anno = clazz.getAnnotation(JmsReceiveMsgAnno.class);

        // 默认的消息监听容器
        DefaultMessageListenerContainer container = new DefaultMessageListenerContainer();
        // 队列连接工厂
        container.setConnectionFactory(applicationContext.getBean(anno.connectionFactory(), ConnectionFactory.class));
        // 监听队列
        container.setDestination(applicationContext.getBean(anno.receiveQueue(), Queue.class));
        // 服务端消费者数量
        container.setConcurrentConsumers(anno.consumers());
        // 消息监听器
        container.setMessageListener(listener);
        // 是否开启事务
        container.setSessionTransacted(anno.sessionTransacted());
        if (StringUtils.isNotEmpty(anno.messageSelector())) {
            // 消息选择器
            container.setMessageSelector(anno.messageSelector());
        }

        messageListenerContainers.add(container);
    }

}

package com.iteye.jms.receive;

import javax.jms.BytesMessage;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;

@JmsReceiveMsgAnno(receiveQueue = "orderReceiveQueue", connectionFactory = "connectionFactoryAdapter")
public class OrderQueueMessageListener implements MessageListener {

    @Override
    public void onMessage(Message message) {
        // 报文消息
        String msgstr = null;

        try {
            if (message instanceof TextMessage) {
                TextMessage tm = (TextMessage) message;
                msgstr = tm.getText();
            } else if (message instanceof BytesMessage) {
                BytesMessage bm = (BytesMessage) message;
                byte[] bys = null;
                bys = new byte[(int) bm.getBodyLength()];
                bm.readBytes(bys);
                msgstr = new String(bys);
            } else {
                // 日志
            }
        } catch (Exception e) {
            // 日志
        }

        System.out.println(msgstr);
    }

}

package com.iteye.jms.receive;

import javax.jms.BytesMessage;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;

@JmsReceiveMsgAnno(receiveQueue = "productReceiveQueue", connectionFactory = "connectionFactoryAdapter")
public class ProductQueueMessageListener implements MessageListener {

    @Override
    public void onMessage(Message message) {
        // 报文消息
        String msgstr = null;

        try {
            if (message instanceof TextMessage) {
                TextMessage tm = (TextMessage) message;
                msgstr = tm.getText();
            } else if (message instanceof BytesMessage) {
                BytesMessage bm = (BytesMessage) message;
                byte[] bys = null;
                bys = new byte[(int) bm.getBodyLength()];
                bm.readBytes(bys);
                msgstr = new String(bys);
            } else {
                // 日志
            }
        } catch (Exception e) {
            // 日志
        }

        System.out.println(msgstr);
    }

}

当向OrderReceive_Queue队列发送消息时会触发订单导入消息监听接口，执行onMessage方法。

当向ProductReceive_Queue队列发送消息时会触发商品导入消息监听接口，执行onMessage方法。