By

文章目录
  1. 1. SpringBoot连接多RabbitMQ源 - 云+社区 - 腾讯云
  2. 2. SpringBoot连接多RabbitMQ源
    1. 2.0.1. 配置
  • 3. rabbitmq
  • 4. 如果exchange根据自身类型和消息routingKey无法找到一个合适的queue存储消息，
  • 5. 那么broker会调用basic.return方法将消息返还给生产者;
  • 6. mandatory标志告诉broker代理服务器至少将消息route到一个队列中，
  • 7. 否则就将消息return给发送者;
  • 8. 1.未送达exchange
  • 9. 2.送达exchange却未送道queue的消息 回调returnCallback.（注意）出现2情况时，publisher-confirms 回调的是true
  • 10. rabbitmq
  • 11. 如果exchange根据自身类型和消息routingKey无法找到一个合适的queue存储消息，
  • 12. 那么broker会调用basic.return方法将消息返还给生产者;
  • 13. mandatory标志告诉broker代理服务器至少将消息route到一个队列中，
  • 14. 否则就将消息return给发送者;
  • 15. 1.未送达exchange
  • 16. 2.送达exchange却未送道queue的消息 回调returnCallback.（注意）出现2情况时，publisher-confirms 回调的是true
    1. 16.0.1. 重写连接工厂
    2. 16.0.2. 创建Exchange、Queue并绑定
    3. 16.0.3. 生产��
    4. 16.0.4. 消费��
    5. 16.0.5. 测试验证
      1. 16.0.5.0.1.

    • SpringBoot连接多RabbitMQ源 - 云+社区 - 腾讯云

    原创

    SpringBoot连接多RabbitMQ源

    在实际开发中，很多场景需要异步处理，这时就需要用到RabbitMQ，而且随着场景的增多程序可能需要连接多个RabbitMQ。SpringBoot本身提供了默认的配置可以快速配置连接RabbitMQ，但是只能连接一个RabbitMQ，当需要连接多个RabbitMQ时，默认的配置就不太适用了，需要单独编写每个连接。

    在SpringBoot框架中，我们常用的两个类一般是：

    RabbitTemplate：作为生产、消费消息使用；

    RabbitAdmin：作为申明、删除交换机和队列，绑定和解绑队列和交换机的绑定关系使用。

    所以我们连接多个RabbitMQ就需要重新建立连接、重新实现这两个类。

    代码如下：

    配置

    application.properties配置文件需要配置两个连接：

    server.port=8080

    rabbitmq

    v2.spring.rabbitmq.host=host

    v2.spring.rabbitmq.port=5672

    v2.spring.rabbitmq.username=username

    v2.spring.rabbitmq.password=password

    v2.spring.rabbitmq.virtual-host=virtual-host

    #consume 手动 ack

    v2.spring.rabbitmq.listener.simple.acknowledge-mode=manual

    #1.当mandatory标志位设置为true时，

    如果exchange根据自身类型和消息routingKey无法找到一个合适的queue存储消息，

    那么broker会调用basic.return方法将消息返还给生产者;

    #2.当mandatory设置为false时，出现上述情况broker会直接将消息丢弃;通俗的讲，

    mandatory标志告诉broker代理服务器至少将消息route到一个队列中，

    否则就将消息return给发送者;

    v2.spring.rabbitmq.template.mandatory=true

    #publisher confirms 发送确认

    v2.spring.rabbitmq.publisher-confirms=true

    #returns callback ：

    1.未送达exchange

    2.送达exchange却未送道queue的消息 回调returnCallback.（注意）出现2情况时，publisher-confirms 回调的是true

    v2.spring.rabbitmq.publisher-returns=true

    v2.spring.rabbitmq.listener.simple.prefetch=5

    rabbitmq

    v1.spring.rabbitmq.host=host

    v1.spring.rabbitmq.port=5672

    v1.spring.rabbitmq.username=username

    v1.spring.rabbitmq.password=password

    v1.spring.rabbitmq.virtual-host=virtual-host

    #consume 手动 ack

    v1.spring.rabbitmq.listener.simple.acknowledge-mode=manual

    #1.当mandatory标志位设置为true时，

    如果exchange根据自身类型和消息routingKey无法找到一个合适的queue存储消息，

    那么broker会调用basic.return方法将消息返还给生产者;

    #2.当mandatory设置为false时，出现上述情况broker会直接将消息丢弃;通俗的讲，

    mandatory标志告诉broker代理服务器至少将消息route到一个队列中，

    否则就将消息return给发送者;

    v1.spring.rabbitmq.template.mandatory=true

    #publisher confirms 发送确认

    v1.spring.rabbitmq.publisher-confirms=true

    #returns callback ：

    1.未送达exchange

    2.送达exchange却未送道queue的消息 回调returnCallback.（注意）出现2情况时，publisher-confirms 回调的是true

    v1.spring.rabbitmq.publisher-returns=true

    v1.spring.rabbitmq.listener.simple.prefetch=5

    重写连接工厂

    需要注意的是，在多源的情况下，需要在某个连接加上@Primary注解，表示主连接，默认使用这个连接

    package com.example.config.rabbitmq;

    import com.alibaba.fastjson.JSON;

    import org.springframework.amqp.core.AcknowledgeMode;

    import org.springframework.amqp.core.Message;

    import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;

    import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;

    import org.springframework.amqp.rabbit.connection.ConnectionFactory;

    import org.springframework.amqp.rabbit.connection.CorrelationData;

    import org.springframework.amqp.rabbit.core.RabbitAdmin;

    import org.springframework.amqp.rabbit.core.RabbitTemplate;

    import org.springframework.beans.factory.annotation.Autowired;

    import org.springframework.beans.factory.annotation.Qualifier;

    import org.springframework.beans.factory.annotation.Value;

    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;

    /**

    • Created by shuai on 2019/4/23.

    */

    @Configuration

    public class MultipleRabbitMQConfig {

    @Bean(name = “v2ConnectionFactory”)

    public CachingConnectionFactory hospSyncConnectionFactory(

    @Value(“${v2.spring.rabbitmq.host}”) String host,

    @Value(“${v2.spring.rabbitmq.port}”) int port,

    @Value(“${v2.spring.rabbitmq.username}”) String username,

    @Value(“${v2.spring.rabbitmq.password}”) String password,

    @Value(“${v2.spring.rabbitmq.virtual-host}”) String virtualHost,

    @Value(“${v2.spring.rabbitmq.publisher-confirms}”) Boolean publisherConfirms,

    @Value(“${v2.spring.rabbitmq.publisher-returns}”) Boolean publisherReturns) {

    CachingConnectionFactory connectionFactory = new CachingConnectionFactory();

    connectionFactory.setHost(host);

    connectionFactory.setPort(port);

    connectionFactory.setUsername(username);

    connectionFactory.setPassword(password);

    connectionFactory.setVirtualHost(virtualHost);

    connectionFactory.setPublisherConfirms(publisherConfirms);

    connectionFactory.setPublisherReturns(publisherReturns);

    return connectionFactory;

    }

    @Bean(name = “v2RabbitTemplate”)

    public RabbitTemplate firstRabbitTemplate(

    @Qualifier(“v2ConnectionFactory”) ConnectionFactory connectionFactory,

    @Value(“${v2.spring.rabbitmq.template.mandatory}”) Boolean mandatory) {

    RabbitTemplate v2RabbitTemplate = new RabbitTemplate(connectionFactory);

    v2RabbitTemplate.setMandatory(mandatory);

    v2RabbitTemplate.setConfirmCallback((correlationData, ack, s) -> {

    if (!ack) {

    // LOGGER.info(“{} 发送RabbitMQ消息 ack确认 失败: [{}]”, this.name, JSON.toJSONString(object));

    } else {

    // LOGGER.info(“{} 发送RabbitMQ消息 ack确认 成功: [{}]”, this.name, JSON.toJSONString(object));

    }

    });

    v2RabbitTemplate.setReturnCallback((message, code, s, exchange, routingKey) -> {

    // LOGGER.error(“{} 发送RabbitMQ消息returnedMessage，出现异常，Exchange不存在或发送至Exchange却没有发送到Queue中，message：[{}], code[{}], s[{}], exchange[{}], routingKey[{}]”, new Object[]{this.name, JSON.toJSONString(message), JSON.toJSONString(code), JSON.toJSONString(s), JSON.toJSONString(exchange), JSON.toJSONString(routingKey)});

    });

    return v2RabbitTemplate;

    }

    @Bean(name = “v2ContainerFactory”)

    public SimpleRabbitListenerContainerFactory hospSyncFactory(

    @Qualifier(“v2ConnectionFactory”) ConnectionFactory connectionFactory,

    @Value(“${v2.spring.rabbitmq.listener.simple.acknowledge-mode}”) String acknowledge,

    @Value(“${v2.spring.rabbitmq.listener.simple.prefetch}”) Integer prefetch

    ) {

    SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();

    factory.setConnectionFactory(connectionFactory);

    factory.setAcknowledgeMode(AcknowledgeMode.valueOf(acknowledge.toUpperCase()));

    factory.setPrefetchCount(prefetch);

    return factory;

    }

    @Bean(name = “v2RabbitAdmin”)

    public RabbitAdmin iqianzhanRabbitAdmin(

    @Qualifier(“v2ConnectionFactory”) ConnectionFactory connectionFactory) {

    RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);

    rabbitAdmin.setAutoStartup(true);

    return rabbitAdmin;

    }

    // mq主连接

    @Bean(name = “v1ConnectionFactory”)

    @Primary

    public CachingConnectionFactory publicConnectionFactory(

    @Value(“${v1.spring.rabbitmq.host}”) String host,

    @Value(“${v1.spring.rabbitmq.port}”) int port,

    @Value(“${v1.spring.rabbitmq.username}”) String username,

    @Value(“${v1.spring.rabbitmq.password}”) String password,

    @Value(“${v1.spring.rabbitmq.virtual-host}”) String virtualHost,

    @Value(“${v1.spring.rabbitmq.publisher-confirms}”) Boolean publisherConfirms,

    @Value(“${v1.spring.rabbitmq.publisher-returns}”) Boolean publisherReturns) {

    CachingConnectionFactory connectionFactory = new CachingConnectionFactory();

    connectionFactory.setHost(host);

    connectionFactory.setPort(port);

    connectionFactory.setUsername(username);

    connectionFactory.setPassword(password);

    connectionFactory.setVirtualHost(virtualHost);

    connectionFactory.setPublisherConfirms(publisherConfirms);

    connectionFactory.setPublisherReturns(publisherReturns);

    return connectionFactory;

    }

    @Bean(name = “v1RabbitTemplate”)

    @Primary

    public RabbitTemplate publicRabbitTemplate(

    @Qualifier(“v1ConnectionFactory”) ConnectionFactory connectionFactory,

    @Value(“${v1.spring.rabbitmq.template.mandatory}”) Boolean mandatory) {

    RabbitTemplate v1RabbitTemplate = new RabbitTemplate(connectionFactory);

    v1RabbitTemplate.setMandatory(mandatory);

    v1RabbitTemplate.setConfirmCallback((correlationData, ack, s) -> {

    if (!ack) {

    // LOGGER.info(“{} 发送RabbitMQ消息 ack确认 失败: [{}]”, this.name, JSON.toJSONString(object));

    } else {

    // LOGGER.info(“{} 发送RabbitMQ消息 ack确认 成功: [{}]”, this.name, JSON.toJSONString(object));

    }

    });

    v1RabbitTemplate.setReturnCallback((message, code, s, exchange, routingKey) -> {

    // LOGGER.error(“{} 发送RabbitMQ消息returnedMessage，出现异常，Exchange不存在或发送至Exchange却没有发送到Queue中，message：[{}], code[{}], s[{}], exchange[{}], routingKey[{}]”, new Object[]{this.name, JSON.toJSONString(message), JSON.toJSONString(code), JSON.toJSONString(s), JSON.toJSONString(exchange), JSON.toJSONString(routingKey)});

    });

    return v1RabbitTemplate;

    }

    @Bean(name = “v1ContainerFactory”)

    @Primary

    public SimpleRabbitListenerContainerFactory insMessageListenerContainer(

    @Qualifier(“v1ConnectionFactory”) ConnectionFactory connectionFactory,

    @Value(“${v1.spring.rabbitmq.listener.simple.acknowledge-mode}”) String acknowledge,

    @Value(“${v1.spring.rabbitmq.listener.simple.prefetch}”) Integer prefetch) {

    SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();

    factory.setConnectionFactory(connectionFactory);

    factory.setAcknowledgeMode(AcknowledgeMode.valueOf(acknowledge.toUpperCase()));

    factory.setPrefetchCount(prefetch);

    return factory;

    }

    @Bean(name = “v1RabbitAdmin”)

    @Primary

    public RabbitAdmin publicRabbitAdmin(

    @Qualifier(“v1ConnectionFactory”) ConnectionFactory connectionFactory) {

    RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);

    rabbitAdmin.setAutoStartup(true);

    return rabbitAdmin;

    }

    }

    创建Exchange、Queue并绑定

    再实现RabbitAdmin后，我们就需要根据RabbitAdmin创建对应的交换机和队列，并建立绑定关系

    package com.example.config.rabbitmq;

    import org.springframework.amqp.core.BindingBuilder;

    import org.springframework.amqp.core.Queue;

    import org.springframework.amqp.core.TopicExchange;

    import org.springframework.amqp.rabbit.core.RabbitAdmin;

    import org.springframework.context.annotation.Configuration;

    import javax.annotation.PostConstruct;

    import javax.annotation.Resource;

    /**

    • 创建Queue、Exchange并建立绑定关系

    • Created by shuai on 2019/5/16.

    */

    @Configuration

    public class MyRabbitMQCreateConfig {

    @Resource(name = “v2RabbitAdmin”)

    private RabbitAdmin v2RabbitAdmin;

    @Resource(name = “v1RabbitAdmin”)

    private RabbitAdmin v1RabbitAdmin;

    @PostConstruct

    public void RabbitInit() {

    v2RabbitAdmin.declareExchange(new TopicExchange(“exchange.topic.example.new”, true, false));

    v2RabbitAdmin.declareQueue(new Queue(“queue.example.topic.new”, true));

    v2RabbitAdmin.declareBinding(

    BindingBuilder

    .bind(new Queue(“queue.example.topic.new”, true)) //直接创建队列

    .to(new TopicExchange(“exchange.topic.example.new”, true, false)) //直接创建交换机 建立关联关系

    .with(“routing.key.example.new”)); //指定路由Key

    }

    }

    生产��

    为了后续验证每个连接都建立成功，并且都能生产消息，生产者这里分别使用新生成的RabbitTemplate发送一条消息。

    package com.example.topic;

    import org.springframework.amqp.rabbit.core.RabbitTemplate;

    import org.springframework.beans.factory.annotation.Autowired;

    import org.springframework.stereotype.Component;

    import javax.annotation.Resource;

    @Component

    public class TopicProducer {

    @Resource(name = “v1RabbitTemplate”)

    private RabbitTemplate v1RabbitTemplate;

    @Resource(name = “v2RabbitTemplate”)

    private RabbitTemplate v2RabbitTemplate;

    public void sendMessageByTopic() {

    String content1 = “This is a topic type of the RabbitMQ message example from v1RabbitTemplate”;

    v1RabbitTemplate.convertAndSend(

    “exchange.topic.example.new”,

    “routing.key.example.new”,

    content1);

    String content2 = “This is a topic type of the RabbitMQ message example from v2RabbitTemplate”;

    v2RabbitTemplate.convertAndSend(

    “exchange.topic.example.new”,

    “routing.key.example.new”,

    content2);

    }

    }

    消费��

    这里需要注意在配置消费队列时，需要标识ContainerFactory

    package com.example.topic;

    import org.springframework.amqp.rabbit.annotation.RabbitHandler;

    import org.springframework.amqp.rabbit.annotation.RabbitListener;

    import org.springframework.stereotype.Component;

    @Component

    @RabbitListener(queues = “queue.example.topic.new”, containerFactory = “v2ContainerFactory”)

    public class TopicConsumer {

    @RabbitHandler

    public void consumer(String message) {

    System.out.println(message);

    }

    }

    这样就完成了SpringBoot连接多个RabbitMQ源的示例了，再写一段测试代码验证下。

    测试验证

    package com.example.test;

    import com.example.topic.TopicProducer;

    import org.junit.Test;

    import org.junit.runner.RunWith;

    import org.springframework.beans.factory.annotation.Autowired;

    import org.springframework.boot.test.context.SpringBootTest;

    import org.springframework.test.context.junit4.SpringRunner;

    @RunWith(SpringRunner.class)

    @SpringBootTest

    public class RabbitMQMultipleTest {

    @Autowired

    private TopicProducer topicProducer;

    @Test

    public void topicProducerTest() {

    topicProducer.sendMessageByTopic();

    }

    }

    执行测试代码，验证结果为：

    -155.png

    验证结果

    验证SpringBoot连接多RabbitMQ源成功！

    github地址：Spring Boot 教程、技术栈、示例代码

    登录 后参与评论

    文章目录
    1. 1. SpringBoot连接多RabbitMQ源 - 云+社区 - 腾讯云
    2. 2. SpringBoot连接多RabbitMQ源
      1. 2.0.1. 配置
  • 3. rabbitmq
  • 4. 如果exchange根据自身类型和消息routingKey无法找到一个合适的queue存储消息，
  • 5. 那么broker会调用basic.return方法将消息返还给生产者;
  • 6. mandatory标志告诉broker代理服务器至少将消息route到一个队列中，
  • 7. 否则就将消息return给发送者;
  • 8. 1.未送达exchange
  • 9. 2.送达exchange却未送道queue的消息 回调returnCallback.（注意）出现2情况时，publisher-confirms 回调的是true
  • 10. rabbitmq
  • 11. 如果exchange根据自身类型和消息routingKey无法找到一个合适的queue存储消息，
  • 12. 那么broker会调用basic.return方法将消息返还给生产者;
  • 13. mandatory标志告诉broker代理服务器至少将消息route到一个队列中，
  • 14. 否则就将消息return给发送者;
  • 15. 1.未送达exchange
  • 16. 2.送达exchange却未送道queue的消息 回调returnCallback.（注意）出现2情况时，publisher-confirms 回调的是true
    1. 16.0.1. 重写连接工厂
    2. 16.0.2. 创建Exchange、Queue并绑定
    3. 16.0.3. 生产��
    4. 16.0.4. 消费��
    5. 16.0.5. 测试验证
      1. 16.0.5.0.1.