SpringBoot中的多RabbitMQ数据源配置实现

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

@Data
public abstract class AbstractRabbitConfiguration {
     protected String host;
    protected Integer port;
    protected String userName;
    protected String password;
    protected String virtualHost;
    protected String queueName;
    protected String exchangeName;
    protected String routingKey;
    protected String acknowledge = "manual";
    protected Integer prefetch = 1;
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setHost(host);
        connectionFactory.setPort(port);
        connectionFactory.setVirtualHost(virtualHost);
        connectionFactory.setUsername(userName);
        connectionFactory.setPassword(password);
        connectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
        connectionFactory.setPublisherReturns(Boolean.TRUE);
        return connectionFactory;
    }
}

@Configuration
@ConfigurationProperties(prefix = "kxj.rabbit")
public class RabbitConfig extends AbstractRabbitConfiguration {
    @Bean("primaryConnectionFactory")
    @Primary
    public ConnectionFactory primaryConnectionFactory() {
        return super.connectionFactory();
    }
    @Bean
    @Primary
    public RabbitTemplate rabbitTemplate(@Qualifier("primaryConnectionFactory") ConnectionFactory connectionFactory,
                                         @Qualifier("confirmCallback") ConfirmCallback confirmCallback,
                                         @Qualifier("returnCallback") ReturnCallback returnCallback) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setConfirmCallback(confirmCallback);
        rabbitTemplate.setReturnCallback(returnCallback);
        rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
        return rabbitTemplate;
    }
    @Bean(name = "primaryContainerFactory")
    public SimpleRabbitListenerContainerFactory simpleRabbitListenerContainerFactory(
            SimpleRabbitListenerContainerFactoryConfigurer configurer,
            @Qualifier("primaryConnectionFactory") ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        // 设置ACK确认机制
        factory.setAcknowledgeMode(AcknowledgeMode.valueOf(acknowledge.toUpperCase()));
        // 设置消费者消费条数
        factory.setPrefetchCount(prefetch);
        configurer.configure(factory, connectionFactory);
        return factory;
    }
    @Bean(name = "primaryRabbitAdmin")
    public RabbitAdmin rabbitAdmin(@Qualifier("primaryConnectionFactory") ConnectionFactory connectionFactory) {
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
        rabbitAdmin.setAutoStartup(true);
        // 声明交换机，队列及对应绑定关系
        Queue queue = RabbitmqUtil.createQueue(queueName);
        FanoutExchange exchange = RabbitmqUtil.createFanoutExchange(exchangeName);
        Binding binding = RabbitmqUtil.createBinding(queue, exchange, "");
        RabbitmqUtil.createRabbitAdmin(queue, exchange, binding, rabbitAdmin);
        return rabbitAdmin;
    }
}

@Slf4j
@Component
public class ConfirmCallback implements RabbitTemplate.ConfirmCallback {
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        if (ack) {
            log.info("传递消息到交换机成功,correlationData:{}, cause:{}", JSON.toJSONString(correlationData), cause);
        } else {
            log.error("传递消息到交换机失败,correlationData:{}, cause:{}", JSON.toJSONString(correlationData), cause);
        }
    }
}
@Slf4j
@Component
public class ReturnCallback implements RabbitTemplate.ReturnCallback {
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        String msg = new String(message.getBody());
        log.error(String.format("消息{%s}不能被正确路由，routingKey为{%s}", msg, routingKey));
    }
}

server.port=8895
kxj.rabbit.host=MQ地址
kxj.rabbit.port=MQ端口
kxj.rabbit.virtualHost=/
kxj.rabbit.userName=guest
kxj.rabbit.password=guest
kxj.rabbit.queueName=test.queue
kxj.rabbit.exchangeName=test.exchange
kxj.rabbit.routingKey=test-routing-key

public class RabbitmqUtil {
    public static DirectExchange createDirectExchange(String exchangeName) {
        if (StringUtils.isNotBlank(exchangeName)) {
            return new DirectExchange(exchangeName, true, false);
        }
        return null;
    }
    public static TopicExchange createTopicExchange(String exchangeName) {
        if (StringUtils.isNotBlank(exchangeName)) {
            return new TopicExchange(exchangeName, true, false);
        }
        return null;
    }
    public static FanoutExchange createFanoutExchange(String exchangeName) {
        if (StringUtils.isNotBlank(exchangeName)) {
            return new FanoutExchange(exchangeName, true, false);
        }
        return null;
    }
    public static Queue createQueue(String queueName) {
        if (StringUtils.isNotBlank(queueName)) {
            return new Queue(queueName, true);
        }
        return null;
    }
    public static Binding createBinding(Queue queueName, Exchange exchangeName, String routingKeyName) {
        if (Objects.nonNull(queueName) && Objects.nonNull(exchangeName)) {
            return BindingBuilder.bind(queueName).to(exchangeName).with(routingKeyName).noargs();
        }
        return null;
    }
//    public static void createRabbitAdmin(Queue queue, DirectExchange exchange, Binding binding, RabbitAdmin rabbitAdmin) {
//        rabbitAdmin.declareQueue(queue);
//        rabbitAdmin.declareExchange(exchange);
//        rabbitAdmin.declareBinding(binding);
//    }
    public static void createRabbitAdmin(Queue queue, Exchange exchange, Binding binding, RabbitAdmin rabbitAdmin) {
        if (queue != null) {
            rabbitAdmin.declareQueue(queue);
        }
        if (exchange != null) {
            rabbitAdmin.declareExchange(exchange);
        }
        if (binding != null) {
            rabbitAdmin.declareBinding(binding);
        }
    }
}

@RunWith(SpringRunner.class)
@SpringBootTest
public class RabbitmqTest {
    @Autowired
    @Qualifier("primaryRabbitAdmin")
    private RabbitAdmin primaryRabbitAdmin;
    @Autowired
    @Qualifier("primaryContainerFactory")
    private SimpleRabbitListenerContainerFactory primaryContainerFactory;
    @Autowired
    @Qualifier("primaryConnectionFactory")
    private ConnectionFactory primaryConnectionFactory;
    @Autowired
    private RabbitTemplate primaryRabbitTemplate;
    @Test
    public void testSend() {
        String message = "Hello, World!";
        primaryRabbitTemplate.convertAndSend("test.exchange", "test.routingKey", message);
        String receivedMessage = (String) primaryRabbitTemplate.receiveAndConvert("test.queue");
        assertEquals(message, receivedMessage);
    }
}