[Java知识库]SpringBoot配置多个RabbitMQ源

SpringBoot配置多个RabbitMQ源

1.yml配置文件

spring:
  rabbitmq:
    inner:
      host: ${RABBIT_MQ_HOST:10.1.16.117}
      port: ${RABBIT_MQ_PORT:5672}
      username: mhs_pvg
      password: Pantech123,
      virtual-host: fims.v2
      exchange: ex.mhs_pvg.front.refresh
    outer:
      host: ${RABBIT_MQ_HOST:106.14.65.01}
      port: ${RABBIT_MQ_PORT:5672}
      username: PANTEPORT
      password: n9F5oftJb874KM90
      virtual-host: telegram
      queue: AFTN_PVG_PARSED_PANPORT

2.config文件

前提：需引入Lombok插件

1. AbstractRabbitConfiguration抽象类

	
package com.pantech.messagehandleraftn.config;

import lombok.Data;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;

/**
 * @author hong
 * @date 2022/3/30 17:22
 * @description MQ配置抽象类
 */
@Data
public abstract class AbstractRabbitConfiguration {
    protected String host;
    protected int port;
    protected String username;
    protected String password;
    protected String virtualHost;

    protected ConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setHost(host);
        connectionFactory.setPort(port);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        connectionFactory.setVirtualHost(virtualHost);
        return connectionFactory;
    }
}

2. InnerRabbitMqConfiguration类

package com.pantech.messagehandleraftn.config;

import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @author hong
 * @date 2022/3/30 17:23
 * @description 内部MQ配置
 */
@Configuration
@ConfigurationProperties("spring.rabbitmq.inner")
public class InnerRabbitMqConfiguration extends AbstractRabbitConfiguration{
    /**
    * 获取连接
    */
    @Bean(name = "innerConnectionFactory")
    public ConnectionFactory innerConnectionFactory() {
        return super.connectionFactory();
    }
    /**
    * 获取生产者操作类(无生产者可删除此类)
    */
    @Bean(name = "innerRabbitTemplate")
    public RabbitTemplate innerRabbitTemplate(@Qualifier("innerConnectionFactory") ConnectionFactory connectionFactory) {
        return new RabbitTemplate(connectionFactory);
    }
    /**
    * 获取消费者通道监听类(无消费者可删除此类)
    */
    @Bean(name = "innerFactory")
    public SimpleRabbitListenerContainerFactory innerFactory(SimpleRabbitListenerContainerFactoryConfigurer configurer,
                                                             @Qualifier("innerConnectionFactory") ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        configurer.configure(factory, connectionFactory);
        return factory;
    }
}

3. OuterRabbitMqConfiguration类

package com.pantech.messagehandleraftn.config;

import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;

/**
 * @author hong
 * @date 2022/3/30 17:25
 * @description 外部MQ配置
 */
@Configuration
@ConfigurationProperties("spring.rabbitmq.outer")
public class OuterRabbitMqConfiguration extends AbstractRabbitConfiguration{
    /**
    * 注意添加@Primary表示默认采用哪个MQ配置
    */
    @Bean(name = "outerConnectionFactory")
    @Primary
    public ConnectionFactory outerConnectionFactory() {
        return super.connectionFactory();
    }

    @Bean(name = "outerFactory")
    public SimpleRabbitListenerContainerFactory outerFactory(SimpleRabbitListenerContainerFactoryConfigurer configurer,
                                                             @Qualifier("outerConnectionFactory") ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        configurer.configure(factory, connectionFactory);
        return factory;
    }
}

3.producer

package com.pantech.messagehandleraftn.producer;

import com.alibaba.fastjson.JSON;
import com.pantech.messagehandleraftn.constant.MessageConstant;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author hong
 * @date 2022/3/30 17:38
 * @description RabbitMQ生产者
 */

@Component
public class RabbitMqProducer {
    /**
    * 引入config中的生产者
    */
    private final RabbitTemplate innerRabbitTemplate;
    /**
     * 前端刷新exchange
     */
    @Value("${spring.rabbitmq.inner.exchange}")
    private String frontRefreshExchange;

    public RabbitMqProducer(RabbitTemplate innerRabbitTemplate) {
        this.innerRabbitTemplate = innerRabbitTemplate;
    }

    public void sendToFront() {
        this.innerRabbitTemplate.send(frontRefreshExchange, MessageConstant.FRONT_ROUTING_KEY, "");
    }
}

4.consumer

package com.pantech.messagehandleraftn.consumer;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

/**
 * @author hong
 * @date 2022/3/1 16:48
 * @description 消费者
 */

@Component
@Slf4j
public class Consumer {
    /**
     * 监听消费者,如果不采用默认配置，需要加上container = "beanName"
     */
    @RabbitListener(queues = "${spring.rabbitmq.outer.queue}")
    public void receive(Message message) {
    	// 原消息
        String messageBody = new String(message.getBody());
        // 业务处理
        ...
    }
}