SpringCloud——Config、Bus、Stream

爪哇缪斯

发布于 2023-05-10 09:54:01

9800

发布于 2023-05-10 09:54:01

文章被收录于专栏：爪哇缪斯爪哇缪斯

一、Spring Cloud Config 1.1> 概述

Spring Cloud Config用来为分布式系统中的基础设施和微服务应用提供集中化的外部配置支持。它分为服务端和客户端两个部分。
服务端——spring-cloud-config-server 它作为分布式配置中心，默认通过配置Git地址，来连接配置仓库并为客户端提供配置信息。
客户端——spring-cloud-config-client 通过它来创建客户端，通过指定配置中心来管理应用资源与业务相关的配置内容，并在启动的时候从配置中心获取和加载配置信息。
由于Spring Cloud Config实现的配置中心默认采用Git来存储配置信息，所以使用Spring Cloud Config构建的配置服务器，天然就支持对微服务应用配置信息的版本管理。

1.2> 实战操作

1.2.1> 构建Server端

首先，引入Config端maven依赖，即：spring-cloud-config-server

通过@EnableConfigServer注解开启Config Server的功能

在application.properties中添加配置信息

【解释】

git.uri：配置Git仓库位置。
git.search-paths：配置仓库路径下的相对搜索位置，可以配置多个。
git.username：访问Git仓库的用户名。
git.password：访问Git仓库的密码。

在Gitee上创建配置文件

启动SpringBoot服务，并执行获取配置信息的请求

【解释】

访问配置信息的URL与配置文件的映射关系如下所示：
- /{application}/{profile}[/{label}]
- /{application}-{profile}.yml
- /{label}/{application}-{profile}.yml
- /{application}-{profile}.properties
- /{label}/{application}-{profile}.properties
label：Git的分支名。例如：config-muse
application：配置的应用名。例如：order
profile：配置的环境名。例如：dev

我们从控制台的输出也可以看到，Config Server从Git中获得配置信息后（git clone），会复制一份到本地系统中，然后读取这些内容并返回给微服务应用进行加载，如下所示：

1.2.2> 构建Client端

在依赖中，加入web和Config Client端依赖

【解释】

在Spring Boot 2.4能够直接在application.properties或application.yml文件中使用新的spring.config.import属性。当使用配置中心时，由于SpringCloud 2020.*以后的版本默认禁用了bootstrap，导致读取配置文件时读取不到该属性。解决这个问题的办法，就是在maven中加入spring-cloud-starter-bootstrap依赖。

在bootstrap.properties中添加配置信息

【解释】

此配置文件的名称一定是bootstrap.properties，因为只有这样，config-server中的配置信息才能被正确的加载。

创建ConfigClientController.java

【解释】

此处可以使用@Value注解来获取配置信息，也可以使用Environment来获取配置信息
启动config-client微服务，针对/mysql和/mysqlEnv这两个接口做请求测试

1.3> 特性说明

1.3.1> 基础架构

基础架构图如下所示：

【执行流程】

1> Config Client端启动后，会根据bootstrap.properties配置文件中所配置的application、profile、label，向Config Server请求获取配置信息。
2> Config Server接到Client端的请求后，根据配置文件中的Git配置信息，连接Git仓库，查找Client端需要的配置信息。
3> 找到配置信息之后，通过git clone将配置信息下载到Server端的文件系统中。
4> Config Server创建Spring的ApplicationContext实例，并从Git本地仓库中加载配置文件，最后将这些配置内容读取出来并返回给客户端应用。
5> Client端在获得外部配置文件后加载到客户端的AplicationContext实例，该配置内容的优先级高于客户端Jar包内部的配置内容，所以在Jar包中重复的内容将不在被加载。

1.3.2> 配置Git本地库

在Git的本地库中，创建4个order的配置文件

然后修改配置文件为：git.uri=file://[git本地库文件地址]

请求测试是否ok

查看Server端后台打印日志

1.3.3> Server端连接的安全配置

由于配置中心存储的配置一般来说都是比较敏感的，所以我们最好是对其连接的时候进行安全性的连接校验。实现起来很简单，只需要在Server端引入Spring Security，如下所示：

为Server端添加用户名和密码

为Client端添加用户名和密码

1.3.4> 高可用配置

只需要在多个Server端配置相同的Git仓库，这样所有的配置内容就通过统一的共享文件系统来维护。而Client端请求的时候，只需要将多个Server的上层配置负载均衡设备即可。由于Server也是微服务，所以，也可以通过将多个Server都注册到Eureka，然后Client端通过服务名来进行请求。

启动Eureka Server

a> 配置服务端

在pom中添加Eureka客户端的Maven依赖

在application.properties中配置服务注册中心

在主类中添加@EnableDiscoveryClient注解

启动config-server服务，可以在Eureka上面看到启动后的服务

b> 配置客户端

在pom中添加Eureka客户端的Maven依赖

添加Eureka注册中心的配置

在主类中添加@EnableDiscoveryClient注解

启动config-client服务，可以在Eureka上面看到启动后的服务

发起请求测试

1.3.5> 配置对Config Server的快速失败检测

不启动Config Server，直接启动Client端应用。默认会预先加载很多其他信息，然后再开始连接Config Server进行属性的注入。如下图所示：

可以通过在bootstrap.properties中配置spring.cloud.config.fail-fast=true来打开快速检测。打开后，启动Client端，可以看到后台很快抛出了异常信息。

1.3.6> 动态刷新配置

引入Actuator的Maven依赖

在Client端的配置文件bootstrap.properties中添加actuator配置信息

首先尝试请求/mysql

在Git上修改配置内容，将jdbc前面加个1，此时再次请求/mysql，返回的内容依然是原先的配置内容。由于actuator监控模块，它包含了/refresh接口的实现，所以我们请求一下/actuator/refresh接口，将实现客户端应用配置信息的重新获取与刷新。返回值中已经提示，mysql有改变。如下图所示：

随后，我们再次调用/mysql请求，发现返回的配置信息中，已经包含了最新的内容，即：jdbc前面已经有“1”。

1.4> 总结

通过上面对Spring Config的介绍，大家应该是对它有了一定的了解。并且，在动态刷新配置的介绍中，我们也能猜想到该功能应该是同Git的Web Hook功能有关联的。也就是说，当Git中配置的内容有变化时，就针对配置了actuator并且发送了/refresh请求的客户端实现配置信息的实时更新。
但是，由于/refresh刷新操作只是通知某个服务实例去获取最新配置，而不是刷新所有的服务实例。所以，随着微服务中服务越来越多，如果还是采用这种/refresh刷新来获取最新配置信息的方式，显然是不合适且工作量浩大的。那么针对于这种情况，我们就可以使用Spring Cloud Bus来实现以消息总线的方式进行配置变更的通知，并完成集群上批量配置更新的操作。

二、Spring Cloud Bus

2.1> 概述

什么叫做消息总线

在微服务架构中，构建公用的消息主题并由其他微服务去订阅和消费，从而起到广播通知的作用，那么我们就称之为消息总线。

我们可以通过Spring Cloud Bus非常便捷的搭建消息总线，比如可以与Spring Cloud Config进行结合，实现配置更新的全局广播。

在当前的Spring Cloud Bus中，仅支持RabbitMQ和Kafka，如果我们使用的是本机的MQ，那么我们甚至都不需要做任何配置，只需要引用Bus的Maven依赖就可以了。下面实战部分，我们将以Kafka为例。

2.2> 实战操作（实现Config的全局刷新）

2.2.1> 前期准备工作

安装&启动Zookeeper

安装&启动Kafka

启动Eureka注册中心

2.2.2> confg-server-bus服务端集成

首先，我们引入spring-cloud-starter-bus-kafka的Maven依赖，由于需要刷新端点，所以也需要依赖actuator

在配置文件application.properties中增加config、Eureka、actuator配置信息

在应用类上开启Eureka客户端（@EnableDiscoveryClient）和Config Server端（@EnableConfigServer）

启动config-server-bus服务

启动服务之后，在控制台中会输出Kafka的连接信息，其中“Resetting the last seen epoch of partition springCloudBus-0”显示消息总线采用的是springCloudBus这个Topic。

INFO 4193 --- [pool-7-thread-1] org.apache.kafka.clients.Metadata        : [Consumer clientId=consumer-anonymous.4a8521bd-2d93-4ee8-a7be-6842d8b8a27c-4, groupId=anonymous.4a8521bd-2d93-4ee8-a7be-6842d8b8a27c] Resetting the last seen epoch of partition springCloudBus-0 to 0 since the associated topicId changed from null to z9pN7q99TVq64Bisub2Ofw

我们可以查看Kafka下Topic为springCloudBus的log文件，该文件中保存着消息数据

登录Eureka控制台http://localhost:8000/，查看Server端服务是否注册成功

2.2.3> confg-client-bus客户端集成

首先，我们引入bus kafka的Maven依赖

在配置文件bootstrap.properties中增加config和Eureka的配置信息

在应用类上开启Eureka客户端（@EnableDiscoveryClient）

启动两个Client端，分别为9001和9002

2.2.4> 通过Bus进行配置的全局广播更新

首先，我们先查询配置信息，获得配置信息为：“jdbc:mysql...”

我们在Gitee上修改配置信息，修改为：“1111jdbc:mysql...”

此时，再次请求获取配置信息接口，发现返回的配置信息还是旧的信息

为了能够全局更新配置，我们请求9000这台Config Server的微服务接口，请求路径为/actuator/busrefresh

刷新完毕后，我们分别查看9001和9002，发现已经获取到了最新的配置信息

2.2.5> 触发配置刷新的两种方式

通过向某一个Config Server端发送刷新配置请求，然后该客户端将消息发送给消息总线，然后其他的服务端获取最新配置信息。但是，如果ServiceA宕机或者服务迁移了，则会影响到我们刷新操作了。为了解决这个问题，我们建议将请求发送给Config Server端。

下面的图就是我们采取将配置刷新请求发送到Config Server端的配置方案。针对该方案，我们需要将Config Server中引入Spring Cloud Bus，即：将配置服务端也加入到消息总线中来。通过这种方式，即使ServiceA~C有任何宕机或迁移，都不会影响到我们对配置的刷新操作。

2.3> 原理及源码解析

2.3.1> 消息通知解析

从上面的例子中，我们知道了消息是通过Kafka为“springCloudBus”的Topic发送的。所以，我们可以开启对这个Topic的消费端。当执行修改完配置信息后，执行/actuator/busrefresh请求，我们就会从Kafka中获得如下消息

把从Kafka中获得的消息Json格式化，如下所示：

【解释】下面，我们来详细理解消息中的信息内容：

type：消息的事件类型。包含： RefreshRemoteApplicationEvent（用来刷新配置的事件）AckRemoteApplicationEvent（响应消息己经正确接收的告知消息事件）
timestamp：消息的时间戳
originService：消息的来源服务实例
id：消息的唯一标识
destinationService：消息的目标服务实例。上面示例中的“**”代表了总线上的所有服务实例。如果想要指定服务或是实例，只需要通过使用/actuator/busrefresh/order:9002请求，就可以仅仅刷新order:9002服务上获取的最新配置信息。如下所示：

上面的消息内容是RefreshRemoteApplicationEvent和AckRemoteApplicationEvent类型共有的，下面几个属性是 AckRemoteApplicationEvent独有的，分别表示如下含义：
- ackId：Ack消息对应的消息来源。我们可以看到第一条AckRemoteApplicationEvent的ackId对应了 RefreshRemoteApplicationEvent的id，说明这条Ack是告知该RefreshRemoteApplicationEvent事件的消息已给被收到了。

2.3.2> 事件驱动模型

Spring的事件驱动类型中包含三个基本概念，分别为： 事件——ApplicationEvent 事件监听者——ApplicationListener 事件发布者——ApplicationEventPublisher和ApplicationEventMulticaster

2.3.2.1> ApplicationEvent事件

当我们需要自定义事件的时候，只需要继承ApplicationEvent，比如：RemoteApplicationEvent和RefreshRemotecApplicationEvent等。
通过上面我们对消息内容的分析，我们了解到消息的事件类型为RemoteApplicationEvent的子类： RefreshRemoteApplicationEvent AckRemoteApplicationEvent
那么我们对于源码的解析，就从这两个类开始分析。如下是这两个类的事件关系类图。

ApplicationEvent的继承关系如下所示，其中包含两个属性，分别为timestamp和source

【解释】

‍source：表示源事件对象。
timestamp：用于存储事件发生的时间戳。‍

我们先来看一下抽象类RemoteApplicationEvent，源码如下所示：

【解释】

@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type")当进行序列化时，会使用子类的名称作为type属性的值。比如："type": "RefreshRemoteApplicationEvent"；

RefreshRemoteApplicationEvent事件类是用于远程刷新应用的配置信息。它只是继承了RemoteApplicationEvent，并没有其他内容。所以，我们上面看到的关于"type": "RefreshRemoteApplicationEvent"的json内容中，与RemoteApplicationEvent中包含的属性完全一致。如下所示：

AckRemoteApplicationEvent事件类用于告知某个事件消息已经被接收，通过该消息我们可以监控各个事件消息的响应。由于Ack消息是有一个事件源头的，而每个事件都必须继承RemoteApplicationEvent抽象类，所以event也就被限制为一定是RemoteApplicationEvent的子类了。

2.3.2.2> ApplicationListener事件监听者

类关系图如下所示：

ApplicationListener继承自接口EventListener，而EventListener是一个的空接口，如下所示：

ApplicationListener的onApplicationEvent(E event)方法对于入参是有限制的，即：要求入参类型是ApplicationEvent的子类，也就是说，只针对ApplicationEvent的子类进行监听和处理。

我们先看一下RefreshListener

【解释】

从onApplicationEvent的入参可以看出，它监听的是RefreshRemoteApplicationEvent事件类，并且通过refresh方法进行配置属性的刷新。

我们在看一下EnvironmentChangeListener的实现

【解释】

它针对EnvironmentChangeRemoteApplicationEvent事件的监听类，在处理类中，可以看到它从EnvironmentChangeRemoteApplicationEvent中获取了之前提到的事件中定义的Map对象，然后通过遍历来更新EnvironmentManager中的属性内容。

2.3.2.3> ApplicationEventPublisher&ApplicationEventMulticaster事件发布者

ApplicationEventPublisher接口定义了发布事件的方法，如下所示：

ApplicationEventMulticaster接口定义了新增&删除&事件广播的方法，如下图所示：

ApplicationEventPublisher中的publishEvent方法是由AbstractApplicationContext来实现的，如下所示：

SimpleApplicationEventMulticaster通过getApplicationListeners方法获得ApplicationListener集合，并对其进行遍历调用监听器的onApplicationEvent()方法来对具体事件做出处理操作。

自动配置类

2.3.3> 控制端点 Endpoint

AbstractBusEndpoint作为控制端点的抽象类，包含了两个子类，分别为：

类图关系如下所示：

三、Spring Cloud Stream

3.1> 概述

消息中间件是我们平时在企业级开发中经常使用的中间件，它具有缓存、解耦、削峰等功能，但是市面上消息中间件很多，比如Kafka，RabbitMQ，RocketMQ，ActiveMQ等等，每个种类的MQ在使用上都是不尽相同的，那么如果我们想要将中间件进行更换，也会面临着要重现实现对接中间件的代码。那么，Spring Cloud Stream的诞生，解决了这部分的内容，不过有一点大家需要注意的就是，它现在只支持Kafka和RabbitMQ，那么它还有那么重要吗？如果我们是使用kafka或者RabbitMQ的话，它不仅仅可以极大的简化我们使用这两种MQ，而且如果要将两种MQ进行切换的话，也非常的便捷。下面我们就来了解一下Spring Cloud Stream。
Spring Cloud Stream是用来为微服务应用构建消息驱动能力的框架，它本质上就是整合了Spring Boot和Spring Integration，实现了一套轻量级的消息驱动的微服务框架。

3.2> 简单例子入门

引入Stream Kafka的Maven依赖

创建用于接收来自Kafka消息的消费者SinkReceiver

启动Spring Boot应用后，通过Kafka客户端kafka-console-producer.sh向topic为“input”的主题上发送消息。./kafka-console-producer.sh --topic input --bootstrap-server localhost:9092

我们可以在控制台看到消息已经接收到了，如下所示：

查看所有topic主题列表——./kafka-topics.sh --list --bootstrap-server localhost:9092

3.3> 核心概念说明

3.3.1> @EnableBinding

该注解用来指定一个或多个定义了@Input或@Output注解的接口，以此实现对消息通道（Channel的绑定）。上面例子中的@EnableBinding(Sink.class)绑定了Sink接口，该接口是Spring Cloud Stream中默认实现的对输入消息通过绑定的定义。如下所示：

【解释】

从上面的源码中，我们可以看到，Sink和Source中分别通过@Input和@Output注解定义了输入通道和输出通道，而Processor通过继承Source和Sink的方式同时定义了一个输入通道和一个输出通道。
其中@Input和@Output注解还有一个value属性，该属性可以用来设置消息通道的名称，这里的Sink和Source中指定的消息通道名称分别为input和output。如果我们直接使用这两个注解而没有指定具体的value值，将默认使用方法名作为消息通道的名称。

3.3.2> @StreamListener

该注解主要是定义在方法上，作用是将被修饰的方法注册为消息中间件上数据流的事件监听器，注解中的属性值对应了监听的消息通道名。在上面的例子中，我们通过@StreamListener(Sink.INPUT)注解将receive方法注册为input消息通道的监听处理器，所以当kafka发送消息的时候，receive方法会做出对应的响应动作。

3.3.3> Spring Cloud Stream应用模型

Spring Cloud Stream构建的应用程序与消息中间件之间是通过绑定器Binder相关联的，绑定器对于应用程序而言起到了隔离作用，它使得不同消息中间件的实现细节对应用程序来说是透明的。如下图所示，在应用程序和Binder之间定义了两条输入通过和三条输出通道来传递消息，而绑定器则是作为这些通道和消息中间件之间的桥梁进行通信。

3.4> 注入绑定接口

在完成了消息通道绑定的定义之后，Spring Cloud Stream会为其创建具体的实例，而开发者只需要通过注入的方式来获取这些实例并直接使用即可。就像下面的例子，我们创建一个SinkSender接口，并通过@EnableBinding来执行绑定操作，那么SinkSender实例便会注入到IOC中，我们可以直接通过@Resource或@Autowired来使用SinkSender实例发送消息。
构建SinkSender接口

绑定SinkSender接口

通过SinkSender实例发送消息

执行http://localhost:8080/sendMsg?msg=aaa请求，可以在控制台看到aaa这个消息

3.5> 注入消息通道

由于Spring Cloud Stream会根据绑定接口中的@Input和@Output注解来创建消息通道实例，所以我们也可以通过直接注入的方式来使用消息通道对象。我们也可以通过@Resource("xxx")或者@Autowired@Qualifier("xxx")来指定要使用的注入实例。如下所示：
创建MultiSinkSender接口

在SinkReceiver中绑定MultiSinkSender接口，并监听“output1”和“output2”这两个消息通道

通过指定name，来确定使用哪个消息通道。在下面例子中，通过入参mothod，确定使用output1Sender还是output2Sender

请求：http://localhost:8080/multiSendMsg?msg=aaa&method=1，我们使用的是output1Sender实例

请求：http://localhost:8080/multiSendMsg?msg=aaa&method=2，我们使用的是output2Sender实例

3.6> Spring Integration原生支持

创建待绑定接口IntegrationProcessor，并指定topic为integration

编写接收方SinkIntegrationReceiver，使用原生的@ServiceActivator注解替换为@StreamListener，实现对IntegrationProcessor.TOPIC通道的监听处理

编写发送方SinkIntegrationSender，其中@InboundChannelAdapter注解定义了该方法是对IntegrationProcessor.TOPIC通道的输出绑定，同时使用poller参数将该方法设置为轮询执行，它会以2秒的频率向IntegrationProcessor.TOPIC通道输出当前时间

启动服务，后台会有如下输出：

3.7> 消费组

通常在生产环境中，我们的每个服务都不会以单节点的方式运行，当同一个服务启动多个实例的时候，这些实例会绑定到同一个消息通道的目标主题上。默认情况下，当生产者发出一条消息到绑定通道上，这条消息会产生多个副本被每个消费者实例接收和处理。但是，在有些业务场景下，我们希望生产者产生的消息只被其中一个实例消费，这个时候就需要为这些消费者设置消费组来实现这样的功能。

3.7.1> 生产者

生产者通过配置spring.cloud.stream.bindings.output.destination指定输入通道对应的主题名为greetings，如下所示：

发送消息类ConsumerGroupSender如下所示：

3.7.2> 消费者

消费者通过配置spring.cloud.stream.bindings.input.destination指定输入通道对应的主题名为greetings；通过配置spring.cloud.stream.bindings.input.group指定消费组名称，启动两个服务，server.port分别为8081和8082，但是都配置相同的消费组名称，比如下面都配置消费组为ConsumerGroup-A，，如下所示：