转载

Spring Cloud Bus：消息总线

在上一篇中，我们使用 Spring Cloud Config 来实现了，分布式配置中心以及动态更新，但是仍然有一个问题，每次更新都需要每个 Config Client 手动调用 /refresh 刷新，这太麻烦了，下面我们就用 Spring Cloud Bus 来解决这个问题。

Spring Cloud Bus

Spring Cloud Bus 将分布式系统的结点与轻量级消息链接起来，可以用于广播状态改变（例如配置改变）或者其他管理指令。

也就是说，配置仓库发生改变了，我们可以通过 Config Server 发起广播通知到各个 Config Client 重新获取配置信息。

这里我们使用 Kafka 来作为消息代理。具体的安装可以查看官网文档

在启动 Kafka 时，如果出现 java.net.UnknownHostException 异常，有两种办法：

在 {kafkapath}/config/server.properties 中添加 host.name=192.168.222.130
在 /etc/hosts 文件中添加 127.0.0.1 主机名

基于上一篇文章的例子，我们创建 bus-server 和 bus-client

在 bus-server 中添加依赖：

<dependencies>
    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-config-server</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-starter-bus-kafka</artifactId>
    </dependency>
</dependencies>

其 application.yml 配置如下：

spring:
  application:
    name: config-server # 名称 
  cloud:
    config:
      server:
        git:
          uri: file://Users/yangdd/Documents/code/GitHub/config-repo 
          default-label: config 
    stream:
      kafka: 
        binder:
          zk-nodes: 192.168.222.130:2181
          brokers: 192.168.222.130:9092        
management:
  security:
    enabled: false   
server:
  port: 4001 # 端口

在 bus-client 中引入依赖：

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-starter-config</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-starter-bus-kafka</artifactId>
    </dependency>
</dependencies>

其 bootstrap.yml 如下：

spring:
  application: 
    name: client
  cloud:
    config:
      uri: http://localhost:4001/
      profile: dev
    stream:
      kafka:
        binder:
          zk-nodes: 192.168.222.130:2181
          brokers: 192.168.222.130:9092
        
management:
  security:
    enabled: false
server:
  port: 4002 # 端口

写个测试方法：

@RefreshScope 
@RestController
public class TestController {

    @Value("${info}")
    private String info;

    @RequestMapping(value = "/getInfo")
    public String getInfo() {
        return this.info;
    }

}

依次启动 bus-server 和 bus-client 。

此时，我们可以使用 kafka-topics --list --zookeeper localhost:2181 命令来查看当前 Kafka 中的 Topic，如果项目成功启动则会发现一个名为 springCloudBus 的 Topic,这个就是 Spring Cloud Bus 的。

现在，访问 http://localhost:4002/getInfo 现在是 config-dev-1.0 。

修改配置仓库中 client-dev.yml 文件 info: config-dev-2.0 。

访问 http://localhost:4001/client/dev 发现配置文件已修改，现在通过 Config Server 发送广播，其实就是通过 POST 方式请求 http://localhost:4001/bus/refresh，再次访问 http://localhost:4002/getInfo 现在是 config-dev-2.0 。

Spring Cloud Bus 还支持指定刷新范围，在某些特殊场景（比如：灰度发布），我们希望可以刷新微服务中某个具体实例的配置，则可以通过 /bus/refresh 接口的 destination 来指定刷新的应用，比如： /bus/refresh?destination=customers:9000 则表示实例名 spring.application.name 为 customers ，端口 server.port 为 9000 的应用程序会刷新配置。当然还可以通过 /bus/refresh?destination=customers:** 指定实例名为 customers 的所有应用程序。