Spring EL 表达式隔离不同环境的 RocketMQ

Posted 2021-09-01 carl-zhao

项目之前使用的是 RabbitMQ 作为消息中间件用来解耦服务之间的调用，现在需要对消息中间件进行升级决定采用 RocketMQ。RocketMQ 相比 RabbitMQ 更具有优势，当然大家可以在网上查找到相关资料。这里就不在赘述了。今天主要是讲如何使用 Spring EL 表达式来隔离不同环境的 RocketMQ 的。

因为在非生产环境我们为了优化资源的效果，只部署了一套 RabbitMQ 环境。但是非生产环境有多套环境：dev(开发环境)、test(测试环境)、pre(预生产环境)。所以我们在使用 RabbitMQ 的时候用虚拟主机对于不同的环境的服务进行资源隔离。

每一个RabbitMQ服务器都能创建虚拟的消息服务器，我们称之为虚拟主机（virtual host），简称为vhost。每一个vhost本质上是一个独立的小型RabbitMQ服务器，拥有自己独立的队列、交换器以及绑定关系等待，并且它拥有自己独立的权限。

当我们使用 RocketMQ 的时候，它就没有虚拟主机这个概念，那只有另想它法了。

在使用 RocketMQ 消息队列时，我们可以使用 Topic 来隔离环境。就是在消息发送的时候 Topic 上面带上环境信息，并且在消息消费需要使用 Topic 与消费组都带上环境信息。比如，我们需要发送一个订单消息，我们假设定义一个 Topic 名为 order：

• 在开发环境(dev)，消息发送方发送消息主题为: order-dev，消息消费方配置消息主题为 order-dev，然后配置消费组为 order-consumer-dev。
• 在测试环境(test)，消息发送方发送消息主题为: order-test，消息消费方配置消息主题为 order-test，然后配置消费组为 order-consumer-test。
• 在预生产环境(pre)，消息发送方发送消息主题为: order-pre，消息消费方配置消息主题为 order-pre，然后配置消费组为 order-consumer-pre。
• 在生产环境(pro)，消息发送方发送消息主题为: order，消息消费方配置消息主题为 order，然后配置消费组为 order-consumer。

在我们使用 RocketMQ 的时候通常是使用官方提供的 Spring Boot Stater 包，比如：

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>2.2.0</version>
</dependency>

我们在使用消息发送的时候使用 RocketMQTemplate 来进行消息发送：

rocketMQTemplate.sendAndReceive("主题", message);

这里的消息主题我们可以使用 @Value 然后通过 Spring EL 表达式来根据环境动态来生成需要发送的消息主题。

在消息消费方，我们使用 @RocketMQMessageListener 注解通过 topic 来定义需要消费的主题，consumerGroup 来定义需要消费组。这两个参数都支持 Spring EL 表达式，所以我们之前的方案是可行的。

RocketMqManager.java

public class RocketMqManager implements EnvironmentAware {

    private Environment environment;

    public String getValue(String value) {
        String profile = getProfile();
        if(isProduct(profile)) {
            return value;
        } else {
            return value + "-" + profile;
        }
    }

    @Override
    public void setEnvironment(Environment environment) {
        this.environment = environment;
    }

    private boolean isProduct(String profile) {
        return "pro".equals(profile);
    }

    private String getProfile(){
        String[] activeProfiles = environment.getActiveProfiles();
        if(ArrayUtils.isEmpty(activeProfiles)) {
            throw new RuntimeException();
        }
        return activeProfiles[0];
    }

}

RocketMqManager 是 RocketMQ 工具类，它会通过不同的环境来生成不同的值来满足我们上面说的资源隔离。生产环境就直接返回传入的值，否则的话就在传入的值后面添加上 "-" + 环境。

Topic

@Data
public class Topic {

    @Value("#{rocketMqManager.getValue('order')}")
    private String topic;

}

Topic 类就是以我们想要通过 Spring EL 表达式根据不同的环境动态获取消息主题(ps:消息消费者组也是同样的逻辑)。这里面 Spring EL 表达式通过调用 Spring Bean 使用到了我们之前定义的工具类。

Config.java

@Configuration
public class Config {

    @Bean("rocketMqManager")
    public RocketMqManager rocketMqManager(){
        return new RocketMqManager();
    }

    @Bean("topic")
    public Topic topic(){
        return new Topic();
    }

}

通过 Spring Class 类型的 bean 来定义 bean 方便我们的测试。

Test.java

public class Test {

    public static void main(String[] args) {
        AnnotationConfigApplicationContext applicationContext = new AnnotationConfigApplicationContext();
        applicationContext.getEnvironment().setActiveProfiles("dev");
        applicationContext.register(Config.class);
        applicationContext.refresh();
        Topic topic = applicationContext.getBean(Topic.class);
        System.out.println("当前 Topic 为 : " + topic.getTopic());
    }

}

测试类，首先我们定义我们的环境是开发环境(dev)，运行结果如下：

当我们修改环境为生产环境(pro)，运行结果如下：

这样就达到我们隔离环境的效果了。

参考文章

• 使用SpringEL表达式进行方法调用
• RocketMQ声明topic以及consumergroup根据不同环境隔离