SpringCloud 11 - SpringCloud Bus 消息总线

1. 消息驱动概述

1.1 什么是 SpringCloudStream？

官方定义 Spring Cloud Stream 是一个构建消息驱动微服务的框架。

应用程序通过 inputs 或者 outputs 与 Spring Cloud Stream 中 binder 对象交互。通过我们配置来 binding (绑定)，而 Spring Cloud Stream的 binder 对象负责与消息中间件交互。所以，我们只需要搞清楚如何与 Spring Cloud Stream 交互就可以方便使用消息驱动的方式。

通过使用 Spring Integration 来连接消息代理中间件以实现消息事件驱动。Spring Cloud Stream 为一些供应商的消息中间件产品提供了个性化的自动化配置实现，引用了发布 - 订阅、消费组、分区的三个核心概念。

目前仅支持 RabbitMQ、Kafka。

① 为什么要用 SpringCloud Stream？

SpringCloud 12 - SpringCloud Stream 消息驱动1. 消息驱动概述2. 案例说明

一句话概括：屏蔽底层消息中间件的差异，降低切换成本，统一消息的编程模型

② 官网：https://spring.io/projects/spring-cloud-stream

SpringCloud 12 - SpringCloud Stream 消息驱动1. 消息驱动概述2. 案例说明

Spring Cloud Stream 用于构建与共享消息传递系统连接的高度可扩展的事件驱动型微服务。该框架提供了一个灵活的编程模型，该模型建立在已经建立并熟悉的 Spring 习惯用法和最佳实践的基础上，包括支持持久化放入发布/订阅。

③ Spring Cloud Stream中文指导手册：https://blog.csdn.net/qq_32734365/article/details/81413218#spring-cloud-stream%E4%B8%AD%E6%96%87%E6%8C%87%E5%AF%BC%E6%89%8B%E5%86%8C

https://m.wang1314.com/doc/webapp/topic/20971999.html

1.2 设计思想

① 标准MQ

SpringCloud 12 - SpringCloud Stream 消息驱动1. 消息驱动概述2. 案例说明

生产者/消费者之间靠消息媒介传递信息内容：Message
消息必须走特定的通道：消息通道 MessageChannel
消息通道里的消息如何被消费呢，谁负责收发处理：消息通道 MessageChannel 的子接口 SubscribableChannel，由 MessageHandler 消息处理器所订阅

② 为什么使用Cloud Stream

比方说我们用到了 RabbitMQ 和 Kafka，由于这两个消息中间件的架构上的不同，像 RabbitMQ 有 exchange，kafka 有 Topic 和 Partitions分区，

SpringCloud 12 - SpringCloud Stream 消息驱动1. 消息驱动概述2. 案例说明

这些中间件的差异性导致我们实际开发项目给我们造成了一定的困扰，我们如果用到了两个消息队列的其中一种，后面的业务需求，我想往另外一种消息队列进行迁移，这时候无疑就是一个灾难性的，一大堆东西都要推倒重新做，因为它跟我们的系统耦合了，这时候 SpringCLoud Stream 给我们提供了一种解耦合的方式。

a）stream 为什么可以统一底层差异？

在没有绑定器这个概念的情况下，SpringBoot 应用要直接与消息中间件进行信息交互的时候，由于各消息中间件构建的初衷不同，它们的实现细节上会有较大的差异性。通过定义绑定器作为中间层，完美地实现了应用程序与消息中间件细节之间的隔离。通过向应用程序暴露统一的 Channel 通道，使得应用程序不需要再考虑各种不同的消息中间件实现。

通过定义绑定器 Binder 作为中间层，实现了应用程序与消息中间件细节之间的隔离。

SpringCloud 12 - SpringCloud Stream 消息驱动1. 消息驱动概述2. 案例说明

b）Binder

SpringCloud 12 - SpringCloud Stream 消息驱动1. 消息驱动概述2. 案例说明

通过定义绑定器 Binder 作为中间层，实现了应用程序与消息中间件细节之间的隔离。

INPUT 对应于消费者
OUTPUT 对应于生产者

③ Stream 中的消息通信方式遵循了发布-订阅模式

Topic 主题进行广播：

在 RabbitMQ 就是 Exchange
在 Kafka 中就是 Topic

1.3 Spring Cloud Stream 标准流程套路

SpringCloud 12 - SpringCloud Stream 消息驱动1. 消息驱动概述2. 案例说明

Binder：很方便的连接中间件，屏蔽差异
Channel：通道，是队列 Queue 的一种抽象，在消息通讯系统中就是实现存储和转发的媒介，通过 Channel 对队列进行配置
Source 和 Sink：简单的可以理解为参照对象是 Spring Cloud Stream 自身，从 Stream 发布消息就是输出，接受消息就是输入

1.4 编码API和常用注解

SpringCloud 12 - SpringCloud Stream 消息驱动1. 消息驱动概述2. 案例说明

2. 案例说明

（前提：有 RabbitMQ 环境。）

2.1 消息驱动之生产者：cloud-stream-rabbitmq-provider8801

① 新建 module cloud-stream-rabbitmq-provider8801

② pom

<dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-actuator</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
        </dependency>
        <!--基础配置-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-devtools</artifactId>
            <scope>runtime</scope>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

③ application.yml

SpringCloud 12 - SpringCloud Stream 消息驱动1. 消息驱动概述2. 案例说明

server:
  port: 8801

spring:
  application:
    name: cloud-stream-provider
  cloud:
      stream:
        binders: # 在此处配置要绑定的rabbitmq的服务信息；
          defaultRabbit: # 表示定义的名称，用于于binding整合
            type: rabbit # 消息组件类型
            environment: # 设置rabbitmq的相关的环境配置
              spring:
                rabbitmq:
                  host: localhost
                  port: 5672
                  username: guest
                  password: guest
        bindings: # 服务的整合处理
          output: # 这个名字是一个通道的名称
            destination: studyExchange # 表示要使用的Exchange名称定义
            content-type: application/json # 设置消息类型，本次为json，文本则设置“text/plain”
            binder: defaultRabbit # 设置要绑定的消息服务的具体设置

eureka:
  client: # 客户端进行Eureka注册的配置
    service-url:
      defaultZone: http://localhost:7001/eureka
  instance:
    lease-renewal-interval-in-seconds: 2 # 设置心跳的时间间隔（默认是30秒）
    lease-expiration-duration-in-seconds: 5 # 如果现在超过了5秒的间隔（默认是90秒）
    instance-id: send-8801.com  # 在信息列表时显示主机名称
    prefer-ip-address: true     # 访问的路径变为IP地址

④ 主启动类

package com.janet.springcloud;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

/**
 * @Description TODO
 * @Date 2020/5/19
 * @Author Janet
 */
@SpringBootApplication
public class StreamMQMain8801 {
    public static void main(String[] args) {
        SpringApplication.run(StreamMQMain8801.class,args);
    }
}

⑤ 业务类

a）发送消息接口

package com.janet.springcloud.service;

/**
 * @Description 发送消息接口
 * @Date 2020/5/19 
 * @Author Janet
 */
public interface IMessageProvider {
    public String send();
}

b）发送消息接口实现类

(官网参考：https://cloud.spring.io/spring-cloud-static/spring-cloud-stream-binder-rabbit/3.0.4.RELEASE/reference/html/spring-cloud-stream-binder-rabbit.html#_partitioning_with_the_rabbitmq_binder)

package com.janet.springcloud.service.impl;

import com.janet.springcloud.service.IMessageProvider;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.MessageBuilder;

import javax.annotation.Resource;
import java.util.UUID;

/**
 * @Description 发送消息接口实现类
 * @Date 2020/5/19
 * @Author Janet
 */
@EnableBinding(Source.class) //定义消息的推送管道，指信道channel和exchange绑定到一起
public class MessageProviderImpl implements IMessageProvider {

    @Resource
    private MessageChannel output; //消息发送管道

    @Override
    public String send() {
        String serial = UUID.randomUUID().toString();
        output.send(MessageBuilder.withPayload(serial).build());
        System.out.println("------serial:"+serial);
        return null;
    }
}

c）Controller：

package com.janet.springcloud.controller;

import com.janet.springcloud.service.IMessageProvider;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;

/**
 * @author Janet
 * @Date 2020/6/11
*/
@RestController
public class SendMessageController {
    @Resource
    private IMessageProvider messageProvider;

    @GetMapping(value = "/sendMessage")
    public String SendMessage(){
        return messageProvider.send();
    }
}

⑥ 测试

启动 7001 eureka、启动 rabbitmq、启动 8801、访问

成功启动之后：

(http://localhost:15672/)

SpringCloud 12 - SpringCloud Stream 消息驱动1. 消息驱动概述2. 案例说明

2.2 消息驱动之消费者：cloud-stream-rabbitmq-consumer8802

① 新建module：cloud-stream-rabbitmq-consumer8802

② POM：

<dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-actuator</artifactId>
        </dependency>
        <!--基础配置-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-devtools</artifactId>
            <scope>runtime</scope>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>

③ YML

SpringCloud 12 - SpringCloud Stream 消息驱动1. 消息驱动概述2. 案例说明

（和 8801 的 yml 类似，就是变成了 input, 还有 instance-id 变了）

server:
  port: 8802

spring:
  application:
    name: cloud-stream-consumer
  cloud:
    stream:
      binders: # 在此处配置要绑定的rabbitmq的服务信息；
        defaultRabbit: # 表示定义的名称，用于于binding整合
          type: rabbit # 消息组件类型
          environment: # 设置rabbitmq的相关的环境配置
            spring:
              rabbitmq:
                host: localhost
                port: 5672
                username: guest
                password: guest
      bindings: # 服务的整合处理
        input: # 这个名字是一个通道的名称
          destination: studyExchange # 表示要使用的Exchange名称定义
          content-type: application/json # 设置消息类型，本次为对象json，如果是文本则设置“text/plain”
          binder: defaultRabbit # 设置要绑定的消息服务的具体设置

eureka:
  client: # 客户端进行Eureka注册的配置
    service-url:
      defaultZone: http://localhost:7001/eureka
  instance:
    lease-renewal-interval-in-seconds: 2 # 设置心跳的时间间隔（默认是30秒）
    lease-expiration-duration-in-seconds: 5 # 如果现在超过了5秒的间隔（默认是90秒）
    instance-id: receive-8802.com  # 在信息列表时显示主机名称
    prefer-ip-address: true     # 访问的路径变为IP地址

④ 主启动类

package com.janet.springcloud;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

/**
 * @Description TODO
 * @Author Janet
 * @Date 2020/6/25
 */
@SpringBootApplication
public class StreamMQMain8802 {
    public static void main(String[] args) {
        SpringApplication.run(StreamMQMain8802.class,args);
    }
}

⑤ 业务类

package com.janet.springcloud.controller;

import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;

/**
 * @Description TODO
 * @Author Janet
 * @Date 2020/6/27
 */
@Component
@EnableBinding(Sink.class)
public class ReceiveMessageListenerController {
    @Value("${server.port}")
    private String serverPort;

    @StreamListener(Sink.INPUT)
    public void input(Message<String> message){
        // 用 withPayload 发送，用 getPayload 接收
        System.out.println("消费者 1 号，------------> 接收到的消息"+message.getPayload()+"\t port: "+serverPort);
    }
}

⑥ 测试 8801 发送，8802 接收消息

启动 7001 eureka、启动 rabbitmq、启动 8801、访问 http://localhost:8801/sendMessage ,

SpringCloud 12 - SpringCloud Stream 消息驱动1. 消息驱动概述2. 案例说明

2.3 分组消费与持久化：cloud-stream-rabbitmq-consumer8803

依照 8802，clone 出来一份 8803

2.3.1 启动

RabbitMQ
7001：服务注册
8801：消息生产
8802：消息消费
8803：消息消费

2.3.2 运行后有两个问题

有重复消费问题
消息持久化问题

2.3.3 重复消费

目前是 8802/8803 同时收到了，存在重复消费问题（导致原因：默认分组group是不同的，组流水号不一样，被认为不同组，可以消费） http://localhost:8801/sendMessage

① 如何解决：分组和持久化属性 group

② 生产实际案例

SpringCloud 12 - SpringCloud Stream 消息驱动1. 消息驱动概述2. 案例说明

不同组是可以全面消费的（重复消费），同一组内会发生竞争关系，只有其中一个可以消费。

2.3.4 分组

① 原理：微服务应用放置于同一个 group 中，就能够保证消息只会被其中一个应用消费一次。不同的组是可以消费的，同一个组内会发生竞争关系，只有其中一个可以消费。

② 8802/8803 都变成不同组，group 两个不同

group：janetgroup1，janetgroup2
8802 修改 YML

SpringCloud 12 - SpringCloud Stream 消息驱动1. 消息驱动概述2. 案例说明

8803 修改 YML

SpringCloud 12 - SpringCloud Stream 消息驱动1. 消息驱动概述2. 案例说明

我们自己配置

SpringCloud 12 - SpringCloud Stream 消息驱动1. 消息驱动概述2. 案例说明

分布式微服务应用为了实现高可用和负载均衡，实际上都会部署多个实例，本次案例启动了两个消费微服务（8802 / 8803）

多数情况下，生产者发送消息给某个具体微服务时只希望被消费一次，按照上面我们启动两个应用的例子，虽然它们同属一个应用，但是这个消息出现了被重复消费两次的情况。为了解决这个问题，在 SpringCloud Stream 中提供了消费组的概念。

结论：还是重复消费

SpringCloud 12 - SpringCloud Stream 消息驱动1. 消息驱动概述2. 案例说明

③ 8802/8803 实现了轮询分组，每次只有一个消费者，8801 模块的发的消息只能被 8802 或 8803 其中一个接收到，这样避免了重复消费。

④ 8802/8803 都变成相同组，group 两个相同

group：janetgroup1
8802 修改YML
8803 修改YML
结论：同一个组的多个微服务实例，每次只会有一个拿到

SpringCloud 12 - SpringCloud Stream 消息驱动1. 消息驱动概述2. 案例说明

2.3.5 持久化

通过上述，解决了重复消费问题，再看看持久化。

停止 8802/8803 服务，并去除掉 8802 分组 group：janetgroup1；8803 的分组 group：janetgroup2 没有去掉。让 8801 先发送 4 条消息到 RabbitMQ，先启动8802，无分组属性配置，后台没有打出来消息；再启动8803，无分组属性配置，后台打出来了 MQ 上的 4 条消息。（所以加上分组信息对消息的持久化至关重要）

SpringCloud 13 - SpringCloud Sleuth 分布式链路跟踪

SpringCloud 12 - SpringCloud Stream 消息驱动1. 消息驱动概述2. 案例说明

1. 消息驱动概述

1.1 什么是 SpringCloudStream？

1.2 设计思想

1.3 Spring Cloud Stream 标准流程套路

1.4 编码API和常用注解

2. 案例说明

2.1 消息驱动之生产者：cloud-stream-rabbitmq-provider8801

2.2 消息驱动之消费者：cloud-stream-rabbitmq-consumer8802

2.3 分组消费与持久化：cloud-stream-rabbitmq-consumer8803

2.3.1 启动

2.3.2 运行后有两个问题

2.3.3 重复消费

2.3.4 分组

2.3.5 持久化

继续阅读

SpringCloud 一、Erueka 注册中心，服务治理与服务注册与发现

删除consul上的节点

【延时任务处理、订单失效】RabbitMQ死信队列实现【订单失效】RabbitMQ死信队列实现

Spring常见知识点1. 什么是 Spring 框架?2. 列举一些重要的Spring模块？3. @RestController vs @Controller4. Spring IOC & AOP5. Spring bean6. Spring MVC7. Spring 框架中用到了哪些设计模式？9. JPA10. Spring Security11. Cookie的作用是什么?和Session有什么区别？

rabbitmq-java-client下载到本地缺少AMQP类

一文教你如何解决RabbitMQ队列无消费者

RabbitMQ简单使用设置账号添加Maven依赖Producer案例Consumer案例

Spring Cloud Gateway Nacos 实现动态路由

Nacos常见问题解决方法

RabbitMQ应用问题——消息补偿机制以及幂等性的保证简单介绍RabbitMQ应用问题消息补偿机制消息幂等性保证

RPC和HTTP对比1、 OSI七层网络模型2、优缺点3、总结

Nacos 集群搭建和持久化配置(Linux)

RabbitMQ安装（Docker、Centos、Windows）

SpringCloud 04 - Zookeeper 服务注册与发现

【持续更新】SpringCloud 学习之异常解决办法总结

SpringCloud(四)Ribbon——负载均衡1 ribbon介绍2 实战