天天看點
傳回

消息中間件之ActiveMQ(三)：SpringBoot整合ActiveMQ整合坑測試

廢話少說，上幹貨：

整合

  1. 建立一個springboot的初始化項目

    略（這個可以會）

  2. pom.xml引入如下包：

    注意：我這裡用的是5.15.11，是以引入的包也是要最新的5.15.11

<dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-activemq</artifactId>
            <version>2.2.4.RELEASE</version>
        </dependency>
        <dependency>
            <groupId>org.apache.activemq</groupId>
            <artifactId>activemq-pool</artifactId>
            <version>5.15.11</version>
        </dependency>
  1. application.yml加入如下配置： 
spring:
  activemq:
    # mq通信位址
    broker-url: tcp://39.96.70.45:61616
    # 是否啟用記憶體模式（就是不安裝MQ，項目啟動時同時啟動一個MQ執行個體）
    in-memory: false
    pool:
      # 是否開啟連接配接池，使用ActiveMQ的連接配接池需要引入activemq-pool依賴
      # true時候會報required a bean of type 'org.springframework.jms.core.JmsMessagingTemplate' that could not be found.異常，文章最後有處理方案。
      enabled: false
      # 空閑的連接配接過期時間，預設為30秒
      idle-timeout: 30000
      # 連接配接池最大連接配接數
      max-connections: 10

  1. 啟動類加上如下注解：

    @EnableJms：聲明對 JMS 注解的支援

@EnableJms
@SpringCloudApplication
public class DemoApplication {
	    public static void main(String[] args) {
        SpringApplication.run(DemoApplication .class, args);
    }
}
  1. 建立activeMQ配置類ActiveMqConfig.java 
package com.jifang.netty.config;

import org.apache.activemq.command.ActiveMQQueue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.jms.Queue;

/**
 * ActiveMq配置
 *
 * @author wangdy
 * @date 2020/2/7 15:15
 */
@Configuration
public class ActiveMqConfig {
	/**
	 * 定義存放消息的隊列
	 */
	@Bean
	public Queue queue() {
		return new ActiveMQQueue("chat-queue");
	}
}
  1. 生産和發送消息 
package com.jifang.netty.service.impl;

import cn.hutool.json.JSONUtil;
import com.jifang.netty.enums.MsgActionTypeEnum;
import com.jifang.netty.enums.MsgTypeEnum;
import com.jifang.netty.service.MessageService;
import com.jifang.netty.service.MqMessageService;
import com.jifang.netty.utils.JsonUtil;
import com.jifang.netty.vo.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.annotation.JmsListener;
import org.springframework.jms.core.JmsMessagingTemplate;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.stereotype.Service;

import javax.jms.Queue;

/**
 * @author wangdy
 * @date 2020/2/7 15:21
 */
@Service
public class MqMessageServiceImpl implements MqMessageService {
	@Autowired
	private Queue queue;
	@Autowired
	private JmsMessagingTemplate jmsMessagingTemplate;
	@Autowired
	private MessageService messageService;

	/**
		發送消息
	*/
	@Override
	public void sendMessage(String message) {
		// 方法1：添加消息到消息隊列
		jmsMessagingTemplate.convertAndSend(queue, message);
		// 方法2：這種方式不需要手動建立queue，系統會自行建立名為testQueue的隊列
		// jmsMessagingTemplate.convertAndSend("testQueue", message);
	}

	/**
		消費消息
	*/
	@Override
	@JmsListener(destination = "chat-queue")// 使用JmsListener配置消費者監聽的隊列，其中message是接收到的消息
	public void handleMessage(String message) {
		System.out.println("成功接收：message" + message);
	}
}

坑

上面的設定pool.enabled=true啟動會報錯,通過檢視源碼，發現由于我這裡springboot的版本用的2.1.x，預設使用的是 @ConditionalOnClass({ JmsPoolConnectionFactory.class, PooledObject.class })，源碼如下：

JmsPoolConnectionFactory

@Configuration
@ConditionalOnMissingBean(ConnectionFactory.class)
class ActiveMQConnectionFactoryConfiguration {

    @Configuration
    @ConditionalOnClass(CachingConnectionFactory.class)
    @ConditionalOnProperty(prefix = "spring.activemq.pool", name = "enabled", havingValue = "false", matchIfMissing = true)
    static class SimpleConnectionFactoryConfiguration {

        private final JmsProperties jmsProperties;

        private final ActiveMQProperties properties;

        private final List<ActiveMQConnectionFactoryCustomizer> connectionFactoryCustomizers;

        SimpleConnectionFactoryConfiguration(JmsProperties jmsProperties,
                ActiveMQProperties properties,
                ObjectProvider<ActiveMQConnectionFactoryCustomizer> connectionFactoryCustomizers) {
            this.jmsProperties = jmsProperties;
            this.properties = properties;
            this.connectionFactoryCustomizers = connectionFactoryCustomizers
                    .orderedStream().collect(Collectors.toList());
        }

        @Bean
        @ConditionalOnProperty(prefix = "spring.jms.cache", name = "enabled", havingValue = "true", matchIfMissing = true)
        public CachingConnectionFactory cachingJmsConnectionFactory() {
            JmsProperties.Cache cacheProperties = this.jmsProperties.getCache();
            CachingConnectionFactory connectionFactory = new CachingConnectionFactory(
                    createConnectionFactory());
            connectionFactory.setCacheConsumers(cacheProperties.isConsumers());
            connectionFactory.setCacheProducers(cacheProperties.isProducers());
            connectionFactory.setSessionCacheSize(cacheProperties.getSessionCacheSize());
            return connectionFactory;
        }

        @Bean
        @ConditionalOnProperty(prefix = "spring.jms.cache", name = "enabled", havingValue = "false")
        public ActiveMQConnectionFactory jmsConnectionFactory() {
            return createConnectionFactory();
        }

        private ActiveMQConnectionFactory createConnectionFactory() {
            return new ActiveMQConnectionFactoryFactory(this.properties,
                    this.connectionFactoryCustomizers)
                            .createConnectionFactory(ActiveMQConnectionFactory.class);
        }

    }

 // 注意：此處用的是import org.messaginghub.pooled.jms.JmsPoolConnectionFactory;
    @Configuration
    @ConditionalOnClass({ JmsPoolConnectionFactory.class, PooledObject.class })
    static class PooledConnectionFactoryConfiguration {

        @Bean(destroyMethod = "stop")
        @ConditionalOnProperty(prefix = "spring.activemq.pool", name = "enabled", havingValue = "true", matchIfMissing = false)
        public JmsPoolConnectionFactory pooledJmsConnectionFactory(
                ActiveMQProperties properties,
                ObjectProvider<ActiveMQConnectionFactoryCustomizer> factoryCustomizers) {
            ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactoryFactory(
                    properties,
                    factoryCustomizers.orderedStream().collect(Collectors.toList()))
                            .createConnectionFactory(ActiveMQConnectionFactory.class);
            return new JmsPoolConnectionFactoryFactory(properties.getPool())
                    .createPooledConnectionFactory(connectionFactory);
        }
    }
}

是以解決方案：

替換原來的activemq-pool 改用：pooled-jms

注意：網上很多地方說springboot2.0.x沒有此問題。我沒細研究，感興趣的可以看下源碼。

<!--  <dependency>
       <groupId>org.apache.activemq</groupId>
       <artifactId>activemq-pool</artifactId>
       <version>5.15.11</version>
   </dependency>-->
<dependency>
    <groupId>org.messaginghub</groupId>
    <artifactId>pooled-jms</artifactId>
</dependency>

測試

運作程式，調用方法發送消息，可以看到隊列消息的變化情況，和控制台列印的接受消息的日志。

如下圖：入隊列數和出隊列數都有了

消息中間件之ActiveMQ(三)：SpringBoot整合ActiveMQ整合坑測試
java activemq springboot整合 連接配接池異常
上一篇: springboot整合redis實作緩存
下一篇: python擷取系統目前時間并轉utc時間為絕對秒數,如何在Python中将datetime轉換為UTC時間戳？...

繼續閱讀