SpringCloud Bus 消息总线

一、前言

        接下来是开展一系列的 SpringCloud 的学习之旅，从传统的模块之间调用，一步步的升级为 SpringCloud 模块之间的调用，此篇文章为第八篇，即介绍 Bus 消息总线。

二、概述

2.1 遗留的问题

        在上一篇文章的最后，我们提出了一个不想手动刷新微服务的问题，即想要实现分布式自动刷新配置功能。Spring Cloud Bus 配合 Spring Cloud Config 使用就可以实现配置的动态刷新。

2.2 Bus 是什么

        Spring Cloud Bus 是用来将分布式系统的节点与轻量级消息系统链接起来的框架，它整合了 Java 的事件处理机制和消息中间件的功能。Spring Clud Bus 目前支持 RabbitMQ 和 Kafka 两种中间件。

2.3 Bus 作用

        Spring Cloud Bus 能管理和传播分布式系统间的消息，就像一个分布式执行器，可用于广播状态更改、事件推送等，也可以当作微服务间的通信通道。

2.4 什么是消息总线

        在微服务架构的系统中，通常会使用轻量级的消息代理来构建一个共用的消息主题，并让系统中所有微服务实例都连接上来。由于该主题中产生的消息会被所有实例监听和消费，所以称它为消息总线。在总线上的各个实例，都可以方便地广播一些需要让其他连接在该主题上的实例都知道的消息。

2.5 基本原理

        所有的 ConfigClient 实例都监听 MQ 中同一个 topic（默认是 springCloudBus）。当一个服务刷新数据的时候，它会把这个信息放入到 Topic 中，这样其它监听同一 Topic 的服务就能得到通知，然后去更新自身的配置。

三、RabbitMQ 环境配置

        使用 Spring Cloud Bus 需要安装 rabbitmq，安装教程在这里，安装完毕后启动，登录，效果如下图：

        等到后面的案例搭建完成之后，会自动的生成一个交换机，如下图：

 

四、Bus 动态刷新全局广播

4.1 设计思想

        一共有两种设计思想，第一种是利用消息总线触发一个客户端，另外一种是利用消息总线触发一个服务端。

4.1.1 触发一个客户端

        利用消息总线触发一个客户端 /bus/refresh，而刷新所有客户端的配置，如下图：

4.1.2 触发一个服务端

        利用消息总线触发一个服务端 ConfigServer 的 /bus/refresh 端点，而刷新所有客户端的配置，如下图：

4.1.3 设计选型

        触发一个服务端的架构（4.1.2）显然更合适一些，因为如果选择架构一就会打破了微服务的职责单一性，因为微服务本身是业务模块，它本不应该承担配置刷新的职责。并且破坏了微服务各节点的对等性。还存在一定的局限性。例如，微服务在迁移时，它的网络地址常常会发生变化，此时如果想要做到自动刷新，那就会增加更多的修改。

4.2 新建工程

        为了演示广播效果，增加复杂度，再以 3355 为模板再制作一个 3366，即新创建一个 cloud-config-center-3366 模块，pom.xml 内容如下所示：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><parent><groupId>com.springcloud</groupId><artifactId>SpringCloud</artifactId><version>1.0-SNAPSHOT</version></parent><artifactId>cloud-config-center-3366</artifactId><properties><maven.compiler.source>8</maven.compiler.source><maven.compiler.target>8</maven.compiler.target><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding></properties><dependencies><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-config</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></dependencies>
</project>

        bootstrap.yml 的内容如下所示：

server:port: 3366spring:application:name: config-clientcloud:#Config客户端配置config:label: master #分支名称name: config #配置文件名称profile: dev #读取后缀名称   上述3个综合：master分支上config-dev.yml的配置文件被读取http://config-3344.com:3344/master/config-dev.ymluri: http://localhost:3344 #配置中心地址#服务注册到eureka地址
eureka:client:service-url:defaultZone: http://localhost:7001/eureka# 暴露监控端点
management:endpoints:web:exposure:include: "*"

        主启动类的代码如下所示：

@EnableEurekaClient
@SpringBootApplication
public class ConfigClientMain3366
{public static void main(String[] args){SpringApplication.run(ConfigClientMain3366.class,args);}
}

        业务类 controller 代码如下所示：

package com.springcloud.controller;import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;@RestController
@RefreshScope
public class ConfigClientController {@Value("${server.port}")private String serverPort;@Value("${config.info}")private String configInfo;@GetMapping("/configInfo")public String configInfo(){return "serverPort: "+serverPort+"\t\n\n configInfo: "+configInfo;}
}

4.3 服务端添加消息总线支持

        给 cloud-config-center-3344 配置中心模块服务端添加消息总线支持，pom.xml 添加如下的依赖：

<!--添加消息总线RabbitMQ支持-->
<dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-bus-amqp</artifactId>
</dependency>

        修改 application.yml ，添加 rabbitmq 的相关配置，如下：

server:port: 3344spring:application:name:  cloud-config-center #注册进Eureka服务器的微服务名cloud:config:server:git:uri: https://github.com/BuGeiQianJiuZa/springcloud-config.git #GitHub上面的git仓库名字####搜索目录search-paths:- springcloud-config####读取分支label: master
#rabbitmq相关配置
rabbitmq:host: localhostport: 5672username: guestpassword: guest#服务注册到eureka地址
eureka:client:service-url:defaultZone: http://localhost:7001/eureka##rabbitmq相关配置,暴露bus刷新配置的端点
management:endpoints: #暴露bus刷新配置的端点web:exposure:include: 'bus-refresh'

4.4 客户端添加消息总线支持

        给 cloud-config-client-3355 客户端添加消息总线支持，pom.xml 添加如下的依赖：

<!--添加消息总线RabbitMQ支持-->
<dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-bus-amqp</artifactId>
</dependency>

        修改 bootstrap.yml ，添加 rabbitmq 的相关配置，如下：

server:port: 3355spring:application:name: config-clientcloud:#Config客户端配置config:label: master #分支名称name: config #配置文件名称profile: dev #读取后缀名称   上述3个综合：master分支上config-dev.yml的配置文件被读取uri: http://localhost:3344 #配置中心地址k#rabbitmq相关配置 15672是Web管理界面的端口；5672是MQ访问的端口rabbitmq:host: localhostport: 5672username: guestpassword: guest#服务注册到eureka地址
eureka:client:service-url:defaultZone: http://localhost:7001/eureka
# 暴露监控端点
management:endpoints:web:exposure:include: "*"   # 'refresh'

         给 cloud-config-client-3366 客户端添加消息总线支持，pom.xml 添加如下的依赖：

<!--添加消息总线RabbitMQ支持-->
<dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-bus-amqp</artifactId>
</dependency>

        修改 bootstrap.yml ，添加 rabbitmq 的相关配置，如下：

server:port: 3366spring:application:name: config-clientcloud:#Config客户端配置config:label: master #分支名称name: config #配置文件名称profile: dev #读取后缀名称   上述3个综合：master分支上config-dev.yml的配置文件被读取uri: http://localhost:3344 #配置中心地址k#rabbitmq相关配置 15672是Web管理界面的端口；5672是MQ访问的端口rabbitmq:host: localhostport: 5672username: guestpassword: guest#服务注册到eureka地址
eureka:client:service-url:defaultZone: http://localhost:7001/eureka
# 暴露监控端点
management:endpoints:web:exposure:include: "*"   # 'refresh'

4.5 测试

        分别启动 cloud-eureka-server7001、cloud-config-client-3344、cloud-config-client-3355 和 cloud-config-client-3366，然后在 gitHub 上修改版本信息，如下图：

        然后给服务端发送一次 post 请求：curl -X POST "http://localhost:3344/actuator/bus-refresh"

        输入 http://config-3344.com:3344/config-dev.yml，测试配置中心，如下图：

        输入 http://config-3344.com:3344/config-dev.ymlhttp://localhost:3355/configInfo，测试 3355 客户端，如下图：

​

        输入 http://localhost:3355/configInfo，测试 3366 客户端，如下图：

​

        可以看到，通过这种方式，所有的客户端的配置信息都已经更新了。

五、Bus 动态刷新定点通知

5.1 思想

        现在不想全部通知，只想定点通知，只通知 3355，不想通知 3366，又该怎么办呢？

5.2 解决方案

        指定具体某一个实例生效而不是全部是有一个公式的，即：

http://localhost:配置中心的端口号/actuator/bus-refresh/{destination}

        /bus/refresh 请求不再发送到具体的服务实例上，而是发给 config server 并通过 destination 参数类指定需要更新配置的服务或实例。

5.3 案例

        我们这里以刷新运行在 3355 端口上的 config-client 为例，更新 gitHub 上的 version 版本，如下：

​

        然后执行以下的命令：

curl -X POST "http://localhost:3344/actuator/bus-refresh/config-client:3355"

​

        输入 http://localhost:3355/configInfo，测试 3355 客户端，如下图：

​

        输入  http://localhost:3366/configInfo，测试 3366 客户端，如下图：

​

        可以看到，通过这种方式，只有 3355 的客户端的配置信息更新了。

5.4 总结

        1、ConfigServer 从 gitHub 上面读取配置信息。并在 rabbitmq 上订阅。

        2、ConfigClient 从 ConfigServer上面读取配置信息。并在 rabbitmq 上订阅。

        3、运维人员手动修改远程 gitHub 上的配置信息。

        4、手动给 ConfigServer 发送 post 请求，告诉监听器配置信息发生了变化，需要刷新。

        5、ConfigServer 发送需要刷新的消息给 rabbitmq。

        6、ConfigClient 接收到了 rabbitmq 发送的需要刷新的消息。

        7、ConfigClient 重新从 ConfigServer上面读取配置信息。