Springboot 支持多MQ數(shù)據(jù)源

config set maxmemory-policy allkeys-lru
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
            <version>2.2.5-RELEASE</version>
        </dependency>

config set maxmemory-policy allkeys-lru
spring:
  rabbitmq:
    normal:
      host: 192.168.96.8
      port: 5672
      username: guest
      password: guest
      template:
        retry:
          enabled: true   #失敗重試
          initial-interval: 10000ms    #第一次重試間隔時(shí)長
          max-interval: 30000ms        #最大重試間隔時(shí)長
          multiplier: 2    # 下次重試間隔的倍數(shù) 2：重試間隔是上次的2倍
      listener:
        simple:
          acknowledge-mode: manual  # 手動(dòng)確認(rèn)ack
    soft:
      host: 192.168.96.6
      port: 5672
      username: guest
      password: guest
      template:
        retry:
          enabled: true   #失敗重試
          initial-interval: 10000ms    #第一次重試間隔時(shí)長
          max-interval: 30000ms        #最大重試間隔時(shí)長
          multiplier: 2    # 下次重試間隔的倍數(shù) 2：重試間隔是上次的2倍
      listener:
        simple:
          acknowledge-mode: manual  # 手動(dòng)確認(rèn)ack

config set maxmemory-policy allkeys-lru
package com.xiaodeai.iot.parser.mq.conf;
 
import com.rabbitmq.client.Channel;
import com.xiaodeai.common.core.constant.LogTypeEnum;
import com.xiaodeai.common.core.utils.LoggerUtil;
import com.xiaodeai.iot.core.constants.IotConstants;
import com.xiaodeai.iot.core.enums.MqQueue;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.context.annotation.Scope;
 
 
/**
 * rabbitmq 多數(shù)據(jù)源配置類
 **/
@Configuration
public class MultiRabbitMqConfig {
 
    @Bean(name = "normalConnectionFactory")
    @Primary
    public ConnectionFactory normalConnectionFactory(
            @Value("${spring.rabbitmq.normal.host}") String host, @Value("${spring.rabbitmq.normal.port}") int port,
            @Value("${spring.rabbitmq.normal.username}") String username, @Value("${spring.rabbitmq.normal.password}") String password) {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setHost(host);
        connectionFactory.setPort(port);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        return connectionFactory;
    }
 
    @Bean(name = "softConnectionFactory")
    public ConnectionFactory softConnectionFactory(
            @Value("${spring.rabbitmq.soft.host}") String host,
            @Value("${spring.rabbitmq.soft.port}") int port,
            @Value("${spring.rabbitmq.soft.username}") String username,
            @Value("${spring.rabbitmq.soft.password}") String password
    ) {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setHost(host);
        connectionFactory.setPort(port);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        return connectionFactory;
    }
 
    @Bean(name = "normalRabbitTemplate")
    @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)//必須是prototype類型
    @Primary
    public RabbitTemplate normalRabbitTemplate(
            @Qualifier("normalConnectionFactory") ConnectionFactory connectionFactory
    ) {
        RabbitTemplate normalRabbitTemplate = new RabbitTemplate(connectionFactory);
        return normalRabbitTemplate;
    }
 
    @Bean(name = "softRabbitTemplate")
    public RabbitTemplate softRabbitTemplate(
            @Qualifier("softConnectionFactory") ConnectionFactory connectionFactory
    ) {
        RabbitTemplate softRabbitTemplate = new RabbitTemplate(connectionFactory);
        return softRabbitTemplate;
    }
 
    @Bean(name = "normalFactory")
    public SimpleRabbitListenerContainerFactory normalFactory(
            SimpleRabbitListenerContainerFactoryConfigurer configurer,
            @Qualifier("normalConnectionFactory") ConnectionFactory connectionFactory
    ) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        configurer.configure(factory, connectionFactory);
        return factory;
    }
 
    @Bean(name = "softFactory")
    public SimpleRabbitListenerContainerFactory softFactory(
            SimpleRabbitListenerContainerFactoryConfigurer configurer,
            @Qualifier("softConnectionFactory") ConnectionFactory connectionFactory
    ) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        configurer.configure(factory, connectionFactory);
        return factory;
    }
 
    /**  
     * 綁定 queue、exchange、routing
     */
    @Bean
    public Channel runNormalQueue(@Qualifier("normalConnectionFactory") ConnectionFactory connectionFactory) {
        Connection connection = connectionFactory.createConnection();
        Channel channel = connection.createChannel(false);
        try {
            channel.queueDeclare("hello1", true, false, false, null);
            channel.queueDeclare("hello2", true, false, false, null);
            
            channel.queueBind("hello1", "test_exchange", "hello1.#");
            channel.queueBind("hello2", "test_exchange", "hello2.#");
        } catch (Exception e) {
            
        }
 
        return channel;
    }
 
    @Bean
    public Channel runSoftQueue(@Qualifier("softConnectionFactory") ConnectionFactory connectionFactory) {
        Connection connection = connectionFactory.createConnection();
        Channel channel = connection.createChannel(false);
        try {
            channel.queueDeclare("hello3", true, false, false, null);
            channel.queueBind("hello3", "test_exchange_1", "hello3.#");
        } catch (Exception e) {
            e.printStackTrace();
        }
        return channel;
    }}

config set maxmemory-policy allkeys-lru
    @Resource(name = "normalRabbitTemplate")
    private RabbitTemplate normalRabbitTemplate;
 
    @Resource(name = "softRabbitTemplate")
    private RabbitTemplate softRabbitTemplate;
 
    /**
     * rabbitmq 消息推送
     *
     * @param exchange 路由
     * @param routing mq路由
     * @param msg 消息內(nèi)容
     */
    public void sendMsg(String exchange, String routing, Object msg) {
//        normalRabbitTemplate.convertAndSend(exchange, routing, msg);
        softRabbitTemplate.convertAndSend(exchange, routing, msg);
    }