我可以使用 RabbitMQ 和 Spring Boot 开发多租户应用程序吗？

Question

我想在我当前的项目中使用 RabbitMQ 和 Spring Boot 来实现多租户应用程序。目前该应用程序以单租户方式运行。该方法是为每个租户创建单独的虚拟主机。我们面临的问题是如何在单个 Spring Boot 应用程序中读取来自不同不同虚拟主机的消息。

我们正在使用 spring boot spring-boot-starter-amqp。

能否请您告诉我这是否可以完成（如果可以，该怎么做）？任何高级代码都会有用吗？

Answer 1

开机自动配置只能配置一个连接工厂。

您需要自己注册基础设施 bean（连接工厂、模板、容器工厂）。

Answer 2

这里的configTemplateRabbitMQ 只是一个包含用户名、密码和地址的对象。

使用cachingConnectionFactory可以设置所有属性。

您可以动态传递信息。

pom.xml

<dependency>
            <groupId>org.springframework.amqp</groupId>
            <artifactId>spring-rabbit</artifactId>
            <version>2.2.10.RELEASE</version>
</dependency>

代码

public void publish(String message, ConfigTemplateRabbitMQ configTemplateRabbitMQ) {
        logger.info("publishing message: {}", message);

        if (message != null) {
            CachingConnectionFactory connectionFactory = new CachingConnectionFactory();

            connectionFactory.setUsername(configTemplateRabbitMQ.getUsername());
            connectionFactory.setPassword(configTemplateRabbitMQ.getPassword());
            connectionFactory.setAddresses(configTemplateRabbitMQ.getAddresses());

            AmqpTemplate dynamicRabbitTemplate = new RabbitTemplate(connectionFactory);

            dynamicRabbitTemplate.convertAndSend(configTemplateRabbitMQ.getExchangeName(), configTemplateRabbitMQ.getPrefix(), message);
        }
        else
            logger.error("message is empty!");
    }

Answer 3

您可以使用对应的AmqpAdmins 声明多个连接工厂。声明AmqpAdmins 后，您可以将它们绑定到您使用的交换/队列。

这里是一些示例@Configuration 类代码，声明了两个不同虚拟主机上的两个交换。

@Configuration
public class RabbitConfiguration {

    @Bean
    public ConnectionFactory firstVhostConnectionFactory(final RabbitProperties properties, @Value("${application.vhost1}") String firstVhostName) {
        var connFactory = new CachingConnectionFactory();
        connFactory.setHost(properties.getHost());
        connFactory.setPort(properties.getPort());
        connFactory.setUsername(properties.getUsername());
        connFactory.setPassword(properties.getPassword());
        connFactory.setVirtualHost(firstVhostName);

        return connFactory;
    }

    @Bean
    ConnectionFactory secondVhostConnectionFactory(final RabbitProperties properties, @Value("${application.vhost2}") String secondVhostName) {
        var connectionFactory = new CachingConnectionFactory();
        connFactory.setHost(properties.getHost());
        connFactory.setPort(properties.getPort());
        connectionFactory.setVirtualHost(secondVhostName);
        connFactory.setUsername(properties.getUsername());
        connFactory.setPassword(properties.getPassword());

        return connectionFactory;
    }

    @Bean
    AmqpAdmin firstVhostAdmin() {
        var admin = new RabbitAdmin(firstVhostConnectionFactory());
        admin.afterPropertiesSet();
        return admin;
    }

    @Bean
    AmqpAdmin secondVhostAdmin() {
        var admin = new RabbitAdmin(secondVhostConnectionFactory());
        admin.afterPropertiesSet();
        return admin;
    }

    @Bean
    FanoutExchange orderStatusRequestExchange() {
        var exchange = new FanoutExchange("secondVhostExchange", true, false);

        // This exchange will be declared on first vhost
        exchange.setAdminsThatShouldDeclare(firstVhostAdmin());
        return exchange;
    }

    @Bean
    FanoutExchange secondVhostExchange() {
        var exchange = new FanoutExchange("secondVhostExchange", true, false);

        // This exchange will be declared on second vhost
        exchange.setAdminsThatShouldDeclare(secondVhostAdmin());
        return exchange;
    }

    @Bean
    Queue orderStatusRequestExchange() {
        var exchange = new FanoutExchange("secondVhostExchange", true, false);

        // This exchange will be declared on first vhost
        exchange.setAdminsThatShouldDeclare(firstVhostAdmin());
        return exchange;
    }

    @Bean
    Queue secondVhostExchange() {
        var exchange = new FanoutExchange("secondVhostExchange", true, false);

        // This exchange will be declared on second vhost
        exchange.setAdminsThatShouldDeclare(secondVhostAdmin());
        return exchange;
    }

    @Bean
    Queue firstVhostQueue() {
        var queue = new AnonymousQueue();
        queue.setAdminsThatShouldDeclare(firstVhostAdmin());
        return queue;
    }


    @Bean
    Queue secondVhostQueue() {
        var queue = new AnonymousQueue();
        queue.setAdminsThatShouldDeclare(secondVhostAdmin());
        return queue;
    }

    @Bean
    Binding firstVhostBinding(FanoutExchange firstVhostExchange, Queue firstVhostQueue) {
        return BindingBuilder.bind(firstVhostQueue).to(firstVhostExchange);
    }


    @Bean
    Binding secondVhostBinding(FanoutExchange secondVhostExchange, Queue secondVhostQueue) {
        return BindingBuilder.bind(secondVhostQueue).to(secondVhostExchange);
    }
}

之后，您可以在 @RabbitListeners 中使用声明的队列并从两个虚拟主机中获取数据。