[版权申明] 非商业目的注明出处可自由转载
出自：shusheng007

概述

在SpringBoot如何整合RabbitMQ中我们留了一个坑，就是如何使用SpringCloud-Stream 来使用RabbitMQ。看名称就知道这个技术是属于SpringCloud家族的一员，SpringCloud从发家起干的就是提供抽象的活，被Netflix晃了一下后在这条路上更是越走越远。SC的宗旨就是：我们不提供核心技术，我们只提供核心技术的整合。但是不得不说人家做的确实是好...

SC为微服务架构中需要的关键组件均提供了一套声明式的使用方法，用户只要按照SpringCloud的方式使用这个组件即可，至于这套方法下面使用的具体技术却是可以被无感知替换的（此处你心中可以打个小折扣...）。今天要说的SpringCloud-Stream就是为了统一消息中间件的技术，你只需要按照SpringCloud的方式使用消息中间件即可，至于下面用的是RabbitMQ还是Kafka都没关系，因为你不是直接调用他们的API，都被抽象过了。例如你刚开始使用的是Rabbitmq，后来想换Kafka，干过工程的都知道这种情况有点扯，基本很少遇到，但是理论上与实际中还真的是存在的，也许就是因为太难换了所以导致在实际中比较少见，如果可能非常容易更换，说不定这种情况就多啦。

来来来，有国才有家， 你不站岗，我不站岗，谁保卫咱祖国，谁来保卫家...，王二狗咋还唱上了，该上干货了

Stream简介

Spring Cloud Stream 是用来构建消息驱动的微服务程序的一个框架。你又问啥是消息驱动? 少年你自己查查吧，咱这篇入门性文章就不搞的太复杂了，信息量太大容易对小朋友们幼小的心灵造成不可磨灭的伤害...然后就放弃了。现在你只要知道，偶现在要用Rabbitmq，使用SpringCloud-Stream咋弄就完事了。

整体架构

下图是官网上的图，我加了几个箭头。

消息中间件上面有一个binder，应用程序通过绑定这个binder与其建立联系，发送消息时应用程序通过output通道将消息传递给binder，binder再把消息给消息中间件。接收消息时消中间件将消息传递给binder，binder再把消息通过input通道传递给应用程序。

SCS的整体思想是：自己定义了一套接口，然后以此操作各种中间件。然而各种中间件的使用接口肯定是各不相同的，那么问题就转变为如何让中间件适配这些接口呢？对了，写个适配器。 从图中可以看出中间件上层被抽象出了一个Binder，它就是适配器。SCS为我们实现了两个：Rabbitmq和Kafka。 但是要是这两个不能满足你的要求，例如你使用了Rocketmq，那你就去官网或者GitHub上找找，看看有没有别人写好的轮子。

这思想简单吧，百试不爽，当前领域处于蛮荒时期，群雄逐鹿，到处是黑马，那么这种面板性的做法越吃的开。如果整个领域就一老大，人家要你干毛，你抽象了半天，下面就一个能打的，人直接就使用那个老大的api了。

工程核心概念

下面说一下使用过程中要了解的核心概念，了解了这些使用的过程中再也不懵逼了

• Binder

你就认为是消息中间件的适配器，通过它你可以访问到消息中间件

• Channel

连接应用程序与Binder的通道，分为input通道与outpu通道。 in与out是站在应用程序的角度说的，你想消费消息那就使用input通道，如果你想发送消息那么就是要output通道

• Binding

将binder与应用程序绑定的组件，只有绑定上了，那应用程序与消息中间件之间就建立了畅通的通道，然后就可以愉快的发送消息了。

理解了上面的几个概念后其实就可以使用SpringClound-Stream了。

值得注意的是，官方在3.1版本后废弃了使用注解的方案，转而推荐使用Java函数模式的方式。

模式核心概念

这又是个什么东东呢？上面介绍的是具体的整合技术，下面介绍的可以认为是使用任何消息中间件都需要考虑的模式性问题。 如果你觉得理解有困难，可以先跳过这部分

• 持久化订阅-发布（persistent publish-subscribe）

  订阅发布模式，这个在编程领域非常流行，也没啥好说的，可以看看这个文章秒懂设计模式之观察者模式（Observer Pattern）

• 消费组（consumer groups）

这个消费组的概念主要为了解决同一个服务多个实例重复消费消息的问题，如果只有一个服务实例是不存在这个问题的。

如图所示，service这个服务订阅了一个主题，当service这个服务运行了两个实例后，同一条消息就会被这两个服务实例消费掉，这样就重复消费了。怎么解决呢？我们就将这两个实例划归为同一个消费组，那样就只有一个实例消费这个消息了，至于谁消费，这又涉及到了负载均衡的问题了，这里我们不管它。

• 分区(partition)

分区又是什么呢？与消费组类似，只有在同一个服务部署多个实例下才有意义。因为现实中有时会存在这种需求：要求符合某些特征的消息必须由同一个消费者实例来处理。例如有这么一类消息 ，根据type的不同需要由不同的实例处理，vip 必须让service#1实例消费，qds必须由service#2消费。

{
    "type": “vip”,
    "msg": "顾客就是上帝"
}

上面类型的消息必须由service实例1消费

{
    "type": “qds”,
    "msg": "的屌丝者的天下"
}

上面类型的消息必须由service实例2消费

partition就是为了解决此类问题的。

整合rabbitmq

SpringBoot整合第三方技术都是经典三部曲

引入依赖

这个稍微复杂点，因为需要引入spring cloud的依赖，而springcloud 是基于springboot的，他们之间的版本有个对应关系，所以需要根据你的springboot来选择springcloud的版本。

我这里springboot的版本是2.6.3，所以我选择了springcloud的2021.0.1版本，他们的对应关系你可以去官网查看。

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

    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-dependencies</artifactId>
                <version>2021.0.1</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
        </dependencies>
    </dependencyManagement>

这个版本的springcloud-stream的版本是3.2.2，而从3.1以后官方就将基于注解的集成方式给废弃了，推荐使用基于函数模型的方式，所以我们这里也就直接使用最新技术了..

修改配置

spring:
   ...
  cloud:
    stream:
      binders:
        defaultRabbit:
          type: rabbit
          environment:
            spring:
              rabbitmq:
                host: localhost
                port: 5672
                username: guest
                password: guest
      bindings:
        ss007Consumer-in-0:
          binder: defaultRabbit
          destination: ss007-auto-topic
          group: ss007-group
        ss007AutoProducer-out-0:
          binder: defaultRabbit
          destination: ss007-auto-topic

    function:
      definition: ss007Consumer;ss007AutoProducer

如果健康检查报链接错误，加上下面的配置

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest

上面的配置有些地方你暂时看不懂，不要担心，下面会一一解释

具体使用

3.1版本以前是使用基于注解的那一套东西，但3.1版本之后就变了。突然从注解改用函数式集成很多人都懵逼了，包括我在内。经过一番调查总算有点眉目了，来来来...

上面提到的Binder概念没有变化，变得只有通道。以前的input和output这些概念还在，只是以前使用注解实现，现在使用Java函数接口来实现了。

总共包含3类

• java.util.function.Supplier

  负责产生消息，你只要在程序中提供一个Supplier的Bean，然后在配置文件中配置一下，程序默认就每秒产生一条消息到配置的消息队列中，这个频率可以配置

• java.util.function.Function

负责中间转换，这个初学时可以不了解，这里为了概念的完整性提一嘴。例如你通过Supplier产生了消息，你可以将这个消息交给Function，Function处理完了再交给Consumer消费

• java.util.function.Consumer

负责消费消息，你只要在程序中提供一个Consumer的Bean，然后再配置文件中配置一下，程序就会监听那个消息队列并处理消息

除了上面3类还有一个东西比较重要，上面提到的那个Supplier都是自动产生消息，而大部分时候我们是要主动发送消息的，例如某人下单了，然后发个消息给短信服务，让它给用户发短信。这里就要用到一个叫

org.springframework.cloud.stream.function.StreamBridge

的东西了，你就把它理解成xxxTemplate就好了

整体就是这样，talk is cheap show me the code， 让我们实践一下吧。

消息自动生产者Supplier

@Slf4j
@Configuration
public class MsgProducer {
    ...
    private int id = 1;

    @Bean
    public Supplier<Message<MsgData>> ss007AutoProducer() {
        return  new Supplier<Message<MsgData>>() {
            @Override
            public Message<MsgData> get() {
                log.info("发送第{}次条消息：",id);
                return MessageBuilder.withPayload(MsgData.builder()
                        .id(id++)
                        .content("我爱你牛翠华")
                        .build())
                        .build();
            }
        };
    }

配置

  cloud:
    stream:
      bindings:
        #代码中的生产者名称
        ss007AutoProducer-out-0:
          binder: defaultRabbit
          destination: ss007-auto-topic
    function:
      definition: ss007AutoProducer

启动程序，查看输出结果

发送第1次条消息：
...
Attempting to connect to: [localhost:5672]
Created new connection: rabbitConnectionFactory.publisher#7e3f8d51:0/SimpleConnection@73fc8210 [delegate=amqp://guest@127.0.0.1:5672/, localPort= 52910]
发送第2次条消息：
发送第3次条消息：
发送第4次条消息：
发送第5次条消息：

可以看到已经建立连接并以每秒一个的频率发送消息了。

我们也可以打开rabbitmq的后台查看一下，ss007-auto-topic这个Exchange里面也一直在被灌入消息

消息消费者Consumer

写一个配置类，提供Consumer的Bean

@Slf4j
@Configuration
public class MsgConsumer {
    @Bean
    public Consumer<MsgData> ss007AutoConsumer(){
        return new Consumer<MsgData>() {
            @Override
            public void accept(MsgData msgData) {
                log.info("ss007AutoConsumer接到消息：{}",msgData.toString());
            }
        };
    }

修改配置文件

  cloud:
    stream:
      bindings:
        ...
        #代码中的消费者名称，格式[函数名称]-in-[index]
        ss007AutoConsumer-in-0:
          binder: defaultRabbit
          #消息主题（kafka）或者交换器（rabbit）
          destination: ss007-auto-topic
          group: ss007-group
   function:
      definition: ss007AutoConsumer

启动程序，查看输出

发送第1次条消息：
ss007AutoConsumer接到消息：MsgData(id=1, content=我爱你牛翠华)
发送第2次条消息：
ss007AutoConsumer接到消息：MsgData(id=2, content=我爱你牛翠华)
发送第3次条消息：
ss007AutoConsumer接到消息：MsgData(id=3, content=我爱你牛翠华)
...

可见生产者每发一条消息，消费者立马就消费了

我们也可以从rabbitmq的后台查看ss007-auto-topic.ss007-group这个队列的信息，可见一直在处理消息。

如何主动发送消息

上面的消息是每秒钟产生一个，但大部分场景下是由某个事件来触发消息的产生，例如你下了个单，然后就会收到短信提醒。以前我们发送消息使用RabbitTemplate，这里使用 StreamBridge就可以了。

使用StreamBridge既可以向Supplier发送，也可以向Function发送。

• 向Supplier发送

前面我们说过默认情况下，系统会定期产生消息，而现在我们要阻止系统这个行为，改为我们主动发送，那么就需要一个配置

cloud.stream.output-bindings = 你的supplier

原来叫cloud.stream.source，真tm的，还没听过就被废弃了...

然后你就可以使用StreamBridge主动发送消息了。

@RequiredArgsConstructor
@Service
public class SendService {
   private final StreamBridge streamBridgeTemplate;
   public void sendMsg2Sup(String msg,Integer id){
      streamBridgeTemplate.send("ss007AutoProducer-out-0",
              MessageBuilder.withPayload(MsgData.builder()
                      .id(id)
                      .content(msg)
                      .build())
                      .build());
   }
 }

上面的代码就向ss007AutoProducer-out-0这个binding发送了一条消息，完整代码请查看文章最后的源码

转化函数

其实介绍完上面的知识后一般的应用场景就够了，但是咱们前面不是还提了一个 java.util.function.Function吗？它用于它既可以用在生产端，也可以用在消费端。那什么时候用function呢？咱的从它能提供的功能来思考，它就是中间负责转化的一个东西。例如有三个服务，S1,S2,S3。S1丢给S2一个消息，S2也不消费，只是要把它丢给S3，那么S2中就可以使用Function，将接到的消息转化成S3需要的格式，然后丢过去。

总之一个原则，java.util.function.Function 要求必须有输入destination和输出destination，需要你在配置文件里配置。

下面的函数就将自己接收到的消息，从MsgData转化为String然后丢给下一个Consumer去消费

    @Bean
    public Function<Message<MsgData>,Message<String>> ss007Function(){
        return new Function<Message<MsgData>, Message<String>>() {
            @Override
            public Message<String> apply(Message<MsgData> msg) {
                MsgData payload = msg.getPayload();
                return MessageBuilder.withPayload(String.format("王二狗第%d说：%s", payload.getId(),payload.getContent()))
                        .build();
            }
        };
    }

配置

      bindings:
        ...
        #代码中function的名称
        ss007Function-in-0:
          binder: defaultRabbit
          destination: ss007-auto-topic
          group: ss007-group
        ss007Function-out-0:
          binder: defaultRabbit
          destination: ss007-func-topic
       ...

从ss007-auto-topic交换器获得消息，处理后丢到ss007-func-topic交换器里。你可以使用StreamBridge主动向ss007Function-in-0发送消息，ss007Function-in-0也可以被动接收消息。什么意思呢？例如上面我们使用StreamBridge向 ss007AutoProducer-out-0发送了消息，由于这个Supplier的destination是ss007-auto-topic 与ss007Function-in-0的一样，所以我们发送给ss007AutoProducer-out-0的消息就会经过ss007Function的转化了。

向ss007AutoProducer-out-0发送消息

      streamBridgeTemplate.send("ss007AutoProducer-out-0",
              MessageBuilder.withPayload(MsgData.builder()
                      .id(id)
                      .content(msg)
                      .build())
                      .build());

输出：

ss007AutoConsumer接到消息：MsgData(id=1, content=我中意你上官无雪)
ss007FuncConsumer接到消息：王二狗第1说：我中意你上官无雪

可见因为是同一个destination，所以function被默认调用了。

具体实例看源码吧。

高级话题

• 分区（Partitioning）

生产端分区

spring.cloud.stream.bindings.callmeEventSupplier-out-0.producer.partitionKeyExpression=分区表达式
spring.cloud.stream.bindings.callmeEventSupplier-out-0.producer.partitionCount=实例个数

• 消费端路由

使用下面配置开启后，当一个消息来了，你可以根据条件控制使用哪个Consumer来消费

spring.cloud.stream.function.routing.enabled=true

spring.cloud.stream.bindings.functionRouter-in-0.destination=
spring.cloud.stream.bindings.functionRouter-in-0.group=b

spring.cloud.function.routing-expression=( headers['type']==1) ? 'consumer1':'consumer2'

本文中在一个服务中模拟了发送和接收消息的场景，一般情况下都是多个服务之间交互。例如你在下单服务里发送一条消息，你同事在短信服务来接收消息，那样你只写xxx-out-[index] Binding，而你同事写xxx-in-[index] Binding即可。

总结

可见SpringCloud-Stream 在代码中完全屏蔽了具体的消息中间件，需要更换的话只需要修改一下配置文件即可。SCS的这次升级，感觉还不如原来那个基于注解的方案好用呢？搞得太玄幻了，半天搞不明白...

源码

你可以从下面获得完整源码，可随手点个小星星，那样就不怕找不到拉

GitHub源码