返回 导航

SpringBoot / Cloud

hangge.com

消息总线Spring Cloud Bus使用详解2（使用Kafka）

作者：hangge | 2020-10-30 08:10

二、整合 Kafka 实现消息总线

1，消息总线介绍

  • 在微服务架构的系统中，我们通常会使用轻量级的消息代理来构建一个共用的消息主题让系统中所有微服务实例都连接上来，由于该主题中产生的消息会被所有实例监听和消费所以我们称它为消息总线。
  • 在总线上的各个实例都可以方便地广播一些需要让其他连接在该主题上的实例都知道的消息，例如配置信息的变更或者其他一些管理操作等。
  • 消息总线在微服务架构系统中被广泛使用，所以它同配置中心一样，几乎是微服务架构中的必备组件。

2，Spring Cloud Bus 介绍

  • Spring Cloud 作为微服务架构综合性的解决方案，也有自己的消息总线实现方案：Spring Cloud Bus。
  • 通过使用 Spring Cloud Bus,可以非常容易地搭建起消息总线，同时实现了一些消息总线中的常用功能。比如：配合 Spring Cloud Config 实现微服务应用配置信息的动态更新等。
  • 当前版本的 Spring Cloud Bus 仅支持两款消息中间件产品：RabbitMQ 和 Kafka。

3，准备工作

（1）在前文中我介绍了如何使用 RabbitMQ 实现消息总线（点击查看），本文接着演示如何使用 Kafka 实现消息总线。所以首次要安装好 Kafka 和 ZooKeeper 环境，具体步骤可以参考我之前写的文章：

（2）本文演示配合 Spring Cloud Config 实现微服务应用配置信息的动态更新，所以需要准备好相关的几个工程：eureka 服务注册中心、配置中心c onfig-server、客户端 hangge-client 以及 Git 仓库中的配置文件。具体内容可以参考我之前写的文章：

4，开始整合

（1）首先修改客户端 hangge-client 的 pom.xml 文件，增加 spring-cloud-starter-bus-kafka 模块。
注意：spring-boot-starter-actuator 模块也是必须的，用来提供刷新端点。 
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-bus-kafka</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-actuator</artifactId>
</dependency>

（2）接着在配置文件中增加关于 Kafka 的连接和用户信息。 
spring.cloud.stream.kafka.binder.brokers=192.168.60.133:9092

5，运行测试

（1）启动 eureka 服务注册中心、配置中心 config-server、以及两个 hangge-client 客户端（分别在不同端口上，比如 7002、7003）

（2）启动后我们通过命令查看 Kafka 中的 Topic，可以发现里面多个了名为 springCloudBus 的 Topic：

（3）分别访问两个客户端的 /test 接口，页面返回对应环境的 from 属性内容。当我们修改 Git 参考里的配置文件后，再次访问 /test 接口，会发现此时属性值并没有变化。

（4）这时我们发送 POST 请求到任意一个客户端实例的 /actuator/bus-refresh 接口上：
（1）这里我们通过向服务实例请求 Spring Cloud Bus 的 /bus-refresh 接口,从而触发总线上其他所有服务实例的 /fresh。当在一些特殊场景下，我们希望可以刷新微服务中某个具体实例的配置，那么可以借助该接口的 destination 参数实现，比如：
  • 请求 /actuator/bus-refresh/hangge-client:7002 则只有实例名为 hangge-client:7002 的应用才会刷新配置。
  • 请求 /actuator/bus-refresh/hangge-client:** 则会触发 hangge-client 服务的所有实例进行刷新。
（2）而从 Git 仓库中配置的修改到发起 /actuator/bus-refresh 的 POST 请求这一步实际中可以同 Git 仓库的 Web Hook 来自动触发。
  • 由于所有连接到消息总线上的应用都会接收到更新请求，所以在 Web Hook 中就不需要维护所有检点内容来进行更新，解决原先每个实例都要通过 Web Hook 来逐个进行刷新的问题。

（5）然后再次访问两个客户端的 /test 接口，可发现两个请求都会返回最新的 from 属性。说明这两个连接到消息总线上的应用都收到了更新请求，并进行了配置更新。

（6）整个系统的架构图如下下所示：

附：架构优化

1，问题描述

（1）在之前的架构中，服务的配置更新需要通过向具体服务中的某个实例发送请求，再触发对整个服务集群的配置更新。
（2）但仍然有个不足，我们指定的应用实例会不同于集群中的其他应用实例，这样会增加集群内部的复杂度，不利于将来的运维工作。
比如需要对服务实例进行迁移，那么我们不得不修改 Web Hook 中的配置等。所以要尽可能地让服务集群中的各个节点是对等的。

2，解决办法

（1）我们在 Config Server 中也引入 Spring Cloud Bus，参考前面客户端配置部分，将配置服务端 config-server 也加入到消息总线中来。
（2）这样以后 /actuator/bus-refresh 请求不在发送到具体服务实例上，而是发送给 Config Server（如果需要指定服务或者实例进行配置更新同样通过 destination 参数指定）

（2）修改后的架构如下。我们的服务实例不需要再承当触发配置更新的指责。同时，对于 Git 的触发等配置都只需要针对 Config Server 即可，从而简化了集群上的一些维护工作。
评论

全部评论（0）

回到顶部