springboot1.x集成activemq

方法一：采用配置文件进行集成

1 添加依赖 pom.xml

<!-- https://mvnrepository.com/artifact/org.apache.activemq/activemq-core -->
		<dependency>
			<groupId>org.apache.activemq</groupId>
			<artifactId>activemq-core</artifactId>
			<version>5.7.0</version>
		</dependency>
		<dependency>
			<groupId>org.apache.activemq</groupId>
			<artifactId>activemq-pool</artifactId>
		</dependency>

2 新增配置文件mq.properties

#设置mq服务器提供服务的ip
broker_url=failover:(tcp://ip:port)

#activeMQ用户名，根据实际情况配置
user= admin
#activeMQ密码，根据实际情况配置
password= admin

#设置发送消息队列主题
send_topic_name=message-server-topic
#初始化发送容器中处理任务的线程数
send_init_thread_in_container=5
#当发送容器中的任务到一定阈值时启动新的线程
send_max_tasks_in_queue=20
#设置发送容器中最多线程数
send_max_thread_in_container=10

#设置接收消息队列主题
receive_queue_name=message-server-queue
#初始化接收容器中处理任务的线程数
receive_init_thread_in_container=5
#当接收容器中的任务到一定阈值时启动新的线程
receive_max_tasks_in_queue=20
#设置接收容器中最多线程数
receive_max_thread_in_container=10

3 创建queue && topic的监听器

注意：项目启动时扫描mq的配置

//新建mq的配置为ApachMqConfig,导入配置文件mq.properties

@Configuration
@PropertySource(value="classpath:mq.properties")
public class ApacheMqConfig {
	
	private static final Logger logger = Logger.getLogger(ApacheMqConfig.class);
	
	@Value("${broker_url}")
	private String brokenUrl;
	
	@Value("${user}")
	private String userName;
	
	@Value("${password}")
	private String password;
	
	@Bean(destroyMethod="stop")
	@ConditionalOnMissingBean
	public PooledConnectionFactory activeMQConnectionFactory(){
		logger.info("apache mq broken url: "+brokenUrl);
		ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(brokenUrl);
		connectionFactory.setOptimizeAcknowledge(true);
		connectionFactory.setOptimizeAcknowledgeTimeOut(400);
		connectionFactory.setUserName(userName);
		connectionFactory.setPassword(password);
		Properties props = new Properties();
		props.setProperty("prefetchPolicy.queuePrefetch","1000");
		props.setProperty("prefetchPolicy.queueBrowserPrefetch","500");
		props.setProperty("prefetchPolicy.durableTopicPrefetch","100");
		props.setProperty("prefetchPolicy.topicPrefetch","400");
		connectionFactory.setProperties(props);
		
		PooledConnectionFactory pooledConnectionFactory = new PooledConnectionFactory();
		pooledConnectionFactory.setConnectionFactory(connectionFactory);
		logger.info("create pooledConnectionFactory: ["+pooledConnectionFactory+"]");
		return pooledConnectionFactory;
	}
	
	/**
	 * JmsListener注解默认只接收queue消息,如果要接收topic消息,需要设置PubSubDomain 为true
	 * @param activeMQConnectionFactory
	 * @return
	 */
	@Bean
    public JmsListenerContainerFactory<?> topicListenerContainer(ConnectionFactory activeMQConnectionFactory) {
        DefaultJmsListenerContainerFactory topicListenerContainer = new DefaultJmsListenerContainerFactory();
        topicListenerContainer.setPubSubDomain(true);
        topicListenerContainer.setConnectionFactory(activeMQConnectionFactory);
        return topicListenerContainer;
    }
	
	/**
	 * 关于消息的发送，是通过 JMS 核心包中的JmsTemplate类来实现的，它简化了 JMS 的使用，
	 * @ConditionalOnMissingBean 该注解表示，如果存在它修饰的类的bean，则不需要再创建这个bean
	 * @param pooledConnectionFactory
	 * @return
	 */
	@Bean
	@ConditionalOnMissingBean
	public JmsTemplate jmsTemplate(PooledConnectionFactory pooledConnectionFactory){
		JmsTemplate jmsTemplate = new JmsTemplate();
		jmsTemplate.setConnectionFactory(pooledConnectionFactory);
		jmsTemplate.setTimeToLive(300000);//有效期5分钟 1000ms*60s*5m
		logger.info("create jmsTemplate: ["+jmsTemplate+"]");
		return jmsTemplate;
	}
}

4 接收mq的queue以及topic，并将xml转换为entity，以及使用

@Component
public class ProducerMq {
  @JmsListener(destination = "xxx.QUEUE")
    public void receiveQueue(String text) throws Exception {
        System.out.println("QUEUE_Consumer收到:"+text);
        
        ReVideoImgEntity reVideoEntity = (ReVideoImgEntity)XmlBuilder.xmlStrToObject(ReVideoImgEntity.class, text);
    }
    
      @JmsListener(destination = "xxx.TOPIC", containerFactory="topicListenerContainer")
    public void receiveTopic(String text) throws Exception {
        System.out.println("TOPIC_Consumer收到:"+text);
      
    	AlarmEntity alermEntity = (AlarmEntity)XmlBuilder.xmlStrToObject(AlarmEntity.class, text);
}

//xml转换为指定类型的pojo
    /**
     * 将XML字符串转换为指定类型的pojo
     * 
     * @param clazz
     * @param xmlStr
     * @return
     */
    public static Object xmlStrToObject(Class<?> clazz, String xmlStr) {
        Object xmlObject = null;
        Reader reader = null;
        try {
            JAXBContext context = JAXBContext.newInstance(clazz);
            // 进行将Xml转成对象的核心接口
            Unmarshaller unmarshaller = context.createUnmarshaller();
            reader = new StringReader(xmlStr);
            xmlObject = unmarshaller.unmarshal(reader);
        } catch (JAXBException e) {
            e.printStackTrace();
        } finally {
            if (null != reader) {
                try {
                    reader.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
        return xmlObject;
    }

方法二：采用配置类进行配置

1 application.properties设置mq参数

 	activemq.broker-url=failover:(tcp://ip:port)?initialReconnectDelay=1000
    activemq.user=admin
    activemq.password= admin 
    activemq.pool.maximumActive=200
    activemq.pool.maxConnections=10
    activemq.pool.idleTimeout=30000

2 编写配置工具类ActiveMQPoolsUtil

package com.njusoft.its.safety.mq;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.pool.PooledConnectionFactory;
import org.springframework.core.env.Environment;

public class ActiveMQPoolsUtil {

    private PooledConnectionFactory pooledConnectionFactory;

   
    public ActiveMQPoolsUtil(Environment env) { 
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory();  
        activeMQConnectionFactory.setUserName(env.getProperty("activemq.user").toString()); 
        activeMQConnectionFactory.setPassword(env.getProperty("activemq.password").toString()); 
        activeMQConnectionFactory.setBrokerURL(env.getProperty("activemq.broker-url").toString()); 
        pooledConnectionFactory= new PooledConnectionFactory(activeMQConnectionFactory); 
        pooledConnectionFactory.setMaximumActiveSessionPerConnection(Integer.valueOf(env.getProperty("activemq.pool.maximumActive"))); 
        pooledConnectionFactory.setIdleTimeout(Integer.valueOf(env.getProperty("activemq.pool.idleTimeout")));  
        pooledConnectionFactory.setMaxConnections(Integer.valueOf(env.getProperty("activemq.pool.maxConnections")));
        pooledConnectionFactory.setBlockIfSessionPoolIsFull(true); 
        System.out.println("---------------已启动MQ");
    } 
    
    
    public PooledConnectionFactory getPooledConnectionFactory() {
        return pooledConnectionFactory;
    }
}

3 编写MQ配置类ActiveMQConfig

package com.njusoft.its.safety.mq;

import javax.jms.ConnectionFactory;
import org.apache.activemq.pool.PooledConnectionFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.Environment;
import org.springframework.jms.annotation.EnableJms;
import org.springframework.jms.config.DefaultJmsListenerContainerFactory;
import org.springframework.jms.config.JmsListenerContainerFactory;

@Configuration
@EnableJms
public class ActiveMQConfig {

    @Autowired
    private Environment env;
    
    /**
     * 主题jms
     * @param activeMQConnectionFactory
     * @return
     */
    @Bean
    public JmsListenerContainerFactory<?> jmsListenerContainerQueue(ConnectionFactory activeMQConnectionFactory) {
    	
    	
    	PooledConnectionFactory pooledConnectionFactory=new ActiveMQPoolsUtil(env).getPooledConnectionFactory();
    	DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        factory.setPubSubDomain(false);
        factory.setConnectionFactory(pooledConnectionFactory);
        return factory;
    }
    /**
     * 主题jms
     * @param activeMQConnectionFactory
     * @return
     */
    @Bean
    public JmsListenerContainerFactory<?> topicListenerContainer(ConnectionFactory activeMQConnectionFactory) {
    	
    	PooledConnectionFactory pooledConnectionFactory=new ActiveMQPoolsUtil(env).getPooledConnectionFactory();
    	DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        factory.setPubSubDomain(true);
        factory.setConnectionFactory(pooledConnectionFactory);
        return factory;
    }

}

4 使用

使用监听器接收消息，主要在于topic消息要使用containerFactory
	/**
	 * 获取图片，视频地址
	 * @param text
	 * @throws Exception 
	 */
	@JmsListener(destination = "xxx.TOPIC", containerFactory="topicListenerContainer")
    public void receiveQueue(String text) throws Exception {
    	LoggerManager.mqInfo("CREATEVIDEO_CALLBACK收到:"+text);
        
        //TODO
    }

两者区别

经试验，第一种配置方法不是很稳定，有时会出现可以接收到消息有时接收不到消息，第二种配置方法目前来看，比较稳定，
不会出现接收不到消息的情况