消息总线Bus

Posted 2021-03-28 Lmg12580

什么是消息总线？

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

Bus是什么？

    Bus是用来将分布式系统的节点与轻量级消息系统链接起来的框架，它整合了Java的事件处理机制和消息中间件的功能。

    Bus支持两种消息代理：RabbitMQ和Kafka

Bus能干什么？

    Bus配合Config使用可以实现配置的动态刷新。

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

基本原理

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

bus+config原理图

接下来围绕这个原理实践一把。

---

一、Bus动态刷新全局广播通知

1. 如上图所示，我们需要多个客户端，所以再以3355为模版，创建一个3366

2. 给3344配置中心服务添加消息总线支持

    a. 增加bus相关依赖包

<dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-bus-amqp</artifactId></dependency>

    b. 修改yml文件，添加如下配置

# rabbitmq相关配置spring: rabbitmq: host: localhost port: 5672 username: guest password: guest# 暴露bus刷新配置的端点management: endpoints: web: exposure: include: "bus-refresh"

3. 客户端3355/3366添加消息总线支持

    a. 添加pom依赖

<dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-bus-amqp</artifactId></dependency>

    b. 修改yml文件，添加如下配置

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

4. 按顺序启动8761/3344/3355/3366四个服务，查看git当前配置文件内容

5. 修改git配置，将version改为5，依次查看服务端及客户端返回内容

6. 向服务端发送POST请求

# curl -X POST "http://localhost:3344/actuator/bus-refresh"

7. 再次查看3344/3355，可以看到在没有重启的情况下，已经取到了最新的配置

一次发送，处处生效

二、Bus动态刷新定点通知

    我们实际工作中还有一种场景：假如我并不想全部通知，我只想让某一个微服务收到通知，这时候就只能用定点通知了。

1. 再次修改git配置文件，版本号改为6，依次查看服务端及客户端返回内容

2. 发送定点通知，指定具体某一个实例生效而不是全部，例如这里我只想通知3355

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

3. 再次查看三个服务的状态，可以看到只有3355取到了最新的配置，3366仍是上一个版本

到这里，Bus的基本知识就学完了

接下来开始学习下个组件：消息驱动Stream

本项目已上传到gitee和github，地址在公众号窗口 我的->git 查看相关内容