Spring Cloud Stream如何屏蔽不同MQ带来的差异性？

本文介绍: Bin der 抽象使 Spr in g Cloud St re a m 应用程序能够灵活地连接中间件。Spring Cloud Strea m 为 Kafk a 和 Ra b bitMQ 提供了 Binder 实现。Roc ketMQ Binder 已由 Spring Cloud Alib a b a 实现。Binder 抽象也是该框架的扩展点之一，我们可以在 Spring Cloud Strea m 之上实现自定义 Binder。

在当前的微服务架构下，使用消息队列（MQ）技术是实现服务解耦和削峰填谷的重要策略。为了保证系统的灵活性和可替换性，我们需要避免对单一开源技术的依赖。

市面上有多种消息队列技术，如 Kafka、Roc ketMQ、Rab bitMQ 等。关键在于如何在微服务体系中实现这些MQ组件的无缝切换，以减少代码修改需求。

Spring Cloud Stream 通过其与主流消息中间件的灵活集成，实现了通过仅修改配置文件的方式来切换不同的MQ实现，从而提高了系统的适应性和可维护性。

Spring Cloud Stream 是一个用于构建消息驱动的微服务应用程序的框架。

基于 Spring Boot 构建，用于创建独立的生产级 Spring 应用程序，并使用 Spring Integration 提供与消息代理的连接。它提供了来自多个供应商的中间件的固定配置，引入了持久发布–订阅语义、消费者组和分区的概念。

简单来说 Spring Cloud Stream 是对 Spring Integration 和 Spring Boot 的合并。

由中间件提供的 Binder 来处理绑定。 应用程序通过绑定这个 Binder 与其建立联系，发送消息时应用程序通过 outputs 通道将消息传递给 Binder，Binder 再把消息给消息中间件。接收消息时消息中间件将消息传递给 Binder，Binder 再把消息通过 inputs 通道传递给应用程序。

pom.xml依赖如下：

消息驱动jar，用哪个mq引入哪个即可。

<dependencies>
    <dependency>
        <groupId>com.alibaba.cloud</groupId>
        <artifactId>spring-cloud-starter-stream-rocketmq</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-starter-stream-kafka</artifactId>
    </dependency>
</dependencies>

<dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-dependencies</artifactId>
            <version>${spring-boot.version}</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
        <dependency>
            <groupId>com.alibaba.cloud</groupId>
            <artifactId>spring-cloud-alibaba-dependencies</artifactId>
            <version>${spring-cloud-alibaba.version}</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-dependencies</artifactId>
            <version>${spring-cloud.version}</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
  </dependencies>
</dependencyManagement>

application.yml RocketMq 配置信息：

spring:
  cloud:
    stream:
      stream:
        rocketmq:
          binder:
            name-server: 127.0.0.1:9876;127.0.0.1:9877
        function:
          # 组装和绑定
          definition: myTopicC
        binders:
          default:
            type: rocketmq
        bindings:
          ## 生产者 新版本固定格式  函数名-{out/in}-{index}
          demoChannel-out-0:
            destination: boot-mq-topic
          ## 消费者 新版本固定格式  函数名字-{out/in}-{index}
          demoChannel-in-0:
            destination: boot-mq-topic

application.yml Kafka 配置信息：

spring:
  cloud:
    stream:
      stream:
        kafka:
          binder:
            brokers: 127.0.0.1:9092
        function:
          # 组装和绑定
          definition: myTopicC
        binders:
          default:
            type: kafka
        bindings:
          ## 生产者 新版本固定格式  函数名-{out/in}-{index}
          demoChannel-out-0:
            destination: boot-mq-topic
          ## 消费者 新版本固定格式  函数名字-{out/in}-{index}
          demoChannel-in-0:
            destination: boot-mq-topic

@RestController
@Slf4j
public class ProducerStream {

    @Autowired
    private StreamBridge streamBridge;

    @GetMapping("/test-stream")
    public String testStream() {
        streamBridge.send("demoChannel-out-0",
                MessageBuilder
                        .withPayload("消息体")
                        .build()
        );
        return "success";
    }
}

@Slf4j
@Configuration
public class TestStreamConsumer {

    @Bean
    public Consumer<String> demoChannel() {
        return message -> {
             log.info("demoChannel接到消息：{}", message);
        };
    }
}

通过图三可以看到 Sping Cloud Stream 的依赖关系。

可以看出来 Sping Cloud Stream 是基于 Spring Integration 做了一层封装，是依赖于 Spring Integration 这个组件的，而 Spring Integration 则依赖于 Spring Messaging 组件来实现消息处理机制的基础设施。

public interface Message<T> {
  //消息体
  T getPayload();

  //消息头
  MessageHeaders getHeaders();
}

@FunctionalInterface
public interface MessageChannel {
 long INDEFINITE_TIMEOUT = -1;
    
    //发送消息，无限期阻塞
 default boolean send(Message<?> message) {
  return send(message, INDEFINITE_TIMEOUT);
 }

    //发送消息，阻塞直到到达指定超时时间
 boolean send(Message<?> message, long timeout);
}

这两个通道都继承自具有消息发送功能的 MessageChannel。

public interface SubscribableChannel extends MessageChannel {
 //通过注册回调函数MessageHandler来实现事件响应
    //注册消息处理器
 boolean subscribe(MessageHandler handler);

 //取消注册消息处理器
 boolean unsubscribe(MessageHandler handler);
}

public interface PollableChannel extends MessageChannel {
 //通过轮询操作主动获取消息
 //从通道中接收消息
 @Nullable
 Message<?> receive();

    //指定超时时间，从通道中接收消息
 @Nullable
 Message<?> receive(long timeout);
}

@FunctionalInterface
public interface MessageHandler {
 //处理消息方法
 void handleMessage(Message<?> message) throws MessagingException;
}

再回到图五流程图中，我们最终可以看到 Kafka 和 RocketMQ 通过继承 AbstractMessageHandler 抽象类（ AbstractMessageHandler 抽象类是实现了 MessageHandler 接口）来实现不同中间件的消息发送操作。而这些都是封装在各自中间件对应的 Binder 代码中来实现。