Spring Cloud Bus 使用说明

约定不等于承诺〃 2022-12-04 07:56 234阅读 0赞

Spring Cloud Bus使用轻量级消息代理将分布式系统的节点链接起来。这代理可以用于广播状态更改（例如配置更改）或其他管理指令。一个关键的想法是，Bus就像一个扩展出去的Spring Boot应用程序的分布式执行器。同时也可以用作应用程序之间的通信渠道。本项目提供AMQP或Kafka作为传输工具。

# 1. 快速开始 #

Spring Boot的自动配置会自动扫描依赖中是否有Spring Cloud Bus。要启用Bus，只要添加依赖`spring-cloud-starter-bus-amqp`或`spring-cloud-starter-bus-kafka`，Spring Cloud负责其余部分。确保代理（RabbitMQ或Kafka）可用和配置好。在本地主机上运行您不用做任何事情。但如果您远程运行，请使用Spring Cloud Connectors或Spring Boot定义的凭据，例如Rabbit：  
**application.yml**

spring:
      rabbitmq:
        host: mybroker.com
        port: 5672
        username: user
        password: secret

总线当前支持向所有监听中的节点发送消息，也可向特定服务（由Eureka定义）所有实例发送消息。/bus/\*路径下有目前有两个http接口。第一个是/bus/env，发送键值对来更新每个节点的Spring Environment。第二个是/bus/refresh，将重新加载每个应用程序的配置，就像他们都调用了/refresh接口一样。

> Bus涵盖了Rabbit和Kafka，因为这是两个最常见的实现。但Spring Cloud Stream非常灵活，binder可以和spring-cloud-bus结合使用。

# 2. Bus接口 #

Spring Cloud Bus 有两个接口，`/actuator/bus-refresh` 和 `/actuator/bus-env`,分别和Spring Cloud的常用actuator接口`/actuator/refresh` 与`/actuator/env`对应。

## 2.1 Refresh接口 ##

`/actuator/bus-refresh`接口清理`RefreshScope`缓存，并重新绑定`@ConfigurationProperties`。

要暴露`/actuator/bus-refresh`接口，需要添加以下配置：

management.endpoints.web.exposure.include=bus-refresh

## 2.2 Env接口 ##

`/actuator/bus-env`会更新所有包含指定键值对的实例。  
要暴露`/actuator/bus-env`接口，需要添加以下配置：

management.endpoints.web.exposure.include=bus-env

`/actuator/bus-env`接口授受的post请求格式如下：

{ 
        "name": "key1",
        "value": "value1"
    }

# 3. 定位实例 #

每个应用实例都会有一个服务ID，它可以通过属性`spring.cloud.bus.id`设置，其值通常是由多个以冒号分隔的id组成。短id的默认值由`spring.application.name`和`server.port`（或者是`spring.application.index`，如果设置了的话）组成。长ID的默认值是app:index:id，其中：

*  app：是vcap.application.name，如果没有设置，则是spring.application.name
 *  index：是vcap.application.instance\_index，如果没设置，依次取spring.application.index、  
    local.server.port、server.port或0
 *  id：是vcap.application.instance\_id，如果没有设置，则取随机值。

HTTP接口可以添加“目的地”路径参数，如`/bus-refresh/customers:9000`，其中`customers:9000`是服务ID。如果ID由Bus上的一个实例拥有，那么它将处理消息，所有其他实例将忽略它。

# 4. 寻址服务的所有实例 #

在Spring PathMatcher（路径分隔符为冒号:）中使用“destination”路径参数来确定实例是否处理该消息。使用上面的示例，“/bus-env/customers:\*\*”将定位“customers”服务的所有实例。

# 5. 应用程序ID必须唯一 #

总线尝试消除一个事件两次，一次从队列中消除，一次从原来的ApplicationEvent中。为此，它会对比发送的服务id和当前的服务ID。如果服多个实例具有相同的ID，则不会处理事件。在本地机器上运行，每个服务将在不同的端口上，端口是服务ID的一部分。Cloud Foundry提供了区分的索引。为了确保服务ID是唯一的，请将spring.application.index设置为服务的每个实例唯一的值。例如，在application.properties中设置spring.application.index=$\{INSTANCE\_INDEX\}（如果使用configserver，请设置bootstrap.properties）。

# 6. 自定义Message Broker #

Spring Cloud Bus使用 Spring Cloud Stream广播消息。因此，为了获取消息流，只需要在类路径中包含您选择的binder实现。有对应AMQP（RabbitMQ）和Kafka的starter，spring-cloud-starter-bus-\[amqp,kafka\]。一般而言，Spring Cloud Stream遵守Spring Boot自动配置约定，因此例如可以使用spring.rabbitmq.\*配置属性更改AMQP代理地址。Spring Cloud Bus在spring.cloud.bus.\*中具有少量本机配置属性（例如，spring.cloud.bus.destination是使用外部中间件的主题的名称）。通常，默认值就足够了。

要更多地了解如何自定义消息代理设置，请参阅Spring Cloud Stream文档。

# 7. 追踪Bus事件 #

Bus事件（RemoteApplicationEvent的子类）可以通过设置spring.cloud.bus.trace.enabled=true进行跟踪。如果这样做，则Spring Boot TraceRepository（如果存在）将显示每个服务实例发送的所有事件和所有的ack。示例（来自/trace接口）：

{ 
      "timestamp": "2015-11-26T10:24:44.411+0000",
      "info": { 
        "signal": "spring.cloud.bus.ack",
        "type": "RefreshRemoteApplicationEvent",
        "id": "c4d374b7-58ea-4928-a312-31984def293b",
        "origin": "stores:8081",
        "destination": "*:**"
      }
      },
      { 
      "timestamp": "2015-11-26T10:24:41.864+0000",
      "info": { 
        "signal": "spring.cloud.bus.sent",
        "type": "RefreshRemoteApplicationEvent",
        "id": "c4d374b7-58ea-4928-a312-31984def293b",
        "origin": "customers:9000",
        "destination": "*:**"
      }
      },
      { 
      "timestamp": "2015-11-26T10:24:41.862+0000",
      "info": { 
        "signal": "spring.cloud.bus.ack",
        "type": "RefreshRemoteApplicationEvent",
        "id": "c4d374b7-58ea-4928-a312-31984def293b",
        "origin": "customers:9000",
        "destination": "*:**"
      }
    }

这个迹象表明RefreshRemoteApplicationEvent已从customers:9000发送，广播到所有服务，并被customers:9000和stores:8081收到（确认）。

要处理ack信号，您可以向您的应用添加AckRemoteApplicationEvent和SentApplicationEvent类型的@EventListener（并启用跟踪）。或者您可以利用TraceRepository并从中挖掘数据。

> 任何Bus应用程序都可以跟踪ack。但有时在一个可以对数据进行更复杂查询的中央服务中这样做是有用的，或将其转发到专门的跟踪服务。

# 8. 广播自己的事件 #

Bus可以携带任何类型为RemoteApplicationEvent的事件。传输格式是JSON，并且反序列化时需要知道使用的是哪些类型。要注册一个新的类型，需要把它放在org.springframework.cloud.bus.event的子包中。

要自定义事件名称，您可以在自定义类上使用@JsonTypeName，或者使用默认策略（使用类的简单名称）。

> 请注意，生产者和消费者都需要访问类定义。

## 8.1. 在自定义包中注册事件 ##

如果您不能或不想为自定义事件使用org.springframework.cloud.bus.event的子包，则必须使用@RemoteApplicationEventScan指定要扫描的包，扫描类型为RemoteApplicationEvent的事件。用@RemoteApplicationEventScan指定的包包括子包。

例如，如果您有一个名为MyEvent的自定义事件：

package com.acme;
    
    public class MyEvent extends RemoteApplicationEvent { 
        ...
    }

您可以通过以下方式注册事件与解串器：

package com.acme;
    
    @Configuration
    @RemoteApplicationEventScan
    public class BusConfiguration { 
        ...

如果没有指定值，将把使用@RemoteApplicationEventScan的类的包进行注册。在这个例子中，将使用BusConfiguration的包com.acme注册。

您还可以使用@RemoteApplicationEventScan上的value，basePackages或basePackageClasses属性明确指定要扫描的软件包。例如：

package com.acme;
    
    @Configuration
    //@RemoteApplicationEventScan({"com.acme", "foo.bar"})
    //@RemoteApplicationEventScan(basePackages = {"com.acme", "foo.bar", "fizz.buzz"})
    @RemoteApplicationEventScan(basePackageClasses = BusConfiguration.class)
    public class BusConfiguration { 
        ...
    }

@RemoteApplicationEventScan的所有示例都是等效的，因为com.acme包将通过在@RemoteApplicationEventScan上明确指定包来注册。

> 请注意，您可以指定要扫描的多个基本软件包。

# 9. 属性列表 #

请查看 [所有属性列表][Link 1]

[Link 1]: https://cloud.spring.io/spring-cloud-bus/2.2.x/reference/html/appendix.html