Spring Cloud Bus 官方文档

素颜马尾好姑娘i 2022-06-02 05:17 226阅读 0赞

前言

Spring Cloud Bus使用轻量级的消息代理连接分布式系统的各个节点。可以用来广播状态改变（例如：配置改变）或者其他管理指令。其中的关键点是，Bus就像一个放大的Spring Boot应用的分布式执行器，但是它也能用作应用间的通讯通道。我们已经提供了使用一个AMQP或者Kafka broker作为传输方式的启动器，其他的传输方式也有同样的基本特性集（这取决于使用何种传输方式）。

1、快速开始

如果在类路径中检测到自己，Spring Cloud Bus会自动添加Spring Boot自动配置并运行。为了启用bus， 你需要做的仅仅是添加  spring-cloud-starter-bus-amqp 或者   spring-cloud-starter-bus-kafka  到你的依赖管理中，Spring Cloud会帮你做剩下的事情。确保代理（RabbitMQ或者Kafka）可用并已经配置好，如果运行在本地，你不需要做任何事情，但如果你使用Spring Cloud Connector远程运行，Spring Boot会依照惯例来定义broker凭据，例如：Rabbit：

application.yml:

spring:

rabbitmq:

host: [ mybroker.com][mybroker.com]

port: 5672

username: user

password: secret

当前bus支持发送消息到所有监听的节点，或者发送到一个特殊服务 （就像Eureka中定义的服务一样） 的所有节点。未来可能会增加更多的选择条件（即，数据中心Y中的服务X的节点，等等）。在  /bus/\*   启动器命名空间中也有很多http端点。当前有两个实现。第一个，  /bus/env  ，发送key/value对来更新每个节点的Spring环境。第二个，  /bus/refresh  ，重新载入每个应用的配置，就像它们被ping他们自己的  /refresh  端点一样。

2、定位实例

HTTP端点接受一个"destination"参数，例如："/bus/refresh?destination=customers:9000"，这表明destination是一个   ApplicationContext  ID。如果这个ID是Bus上的一个实例所有，那么这个实例会处理这条消息，而其他的示例会忽略掉这条消息。Spring Boot在  ContextIdApplicationContextInitializer  中为用户设置了这个ID，这个ID默认是  spring.application.name  、启用的profile和  server.port  的组合。

3、定位一个服务的所有实例

“destination”参数被用于Spring    PathMatcher  （以冒号作为路径分隔符)中来决定实例是否要处理消息。借用上面的例子，"/bus/refresh?destionation=customers:\*\*"会指向"customers"服务的所有实例，而不管设置为  ApplicationContext   ID 的profile和端口号。

4、Application Context ID 必须是唯一的

Bus对于一个事件只处理两次，一次是从源  ApplicationEvent ，另一次是从队列。为此，Bus会根据当前的Application Context ID来检查发送的Application Context ID。如果一个服务的多个实例有相同的Application Context ID，事件将不会被处理。运行在本地机器上的时候，每个服务会监听不同的端口，而端口会作为Application Context ID的一部分。 Cloud Foundry提供了一个索引来进行区分。为了确保 Application Context  ID是唯一的，在服务中，设置 spring.application.index  属性为实例间唯一。例如，在application.properties（或者bootstrap.properties）中设置 spring.application.index=$\{INSTANCE\_INDEX\}  。

5、自定义Message Broker

Spring Cloud Bus使用Spring Cloud Stream来广播消息，所以，为了让消息传播，你只需要在classpath中包含你选择的binder实现。已经有专门用于Bus集成AMQP(RabbitMQ)或Kafka的十分方便的starter了（ spring-cloud-starter-bus-\[amqp,kafka\] ）。一般来说，Spring Cloud Stream 依赖于Spring Boot的自动配置约定来配置消息中间件，所以，AMQP代理地址可以使用  spring.rabbitmq.\*  配置属性来更改。 Spring Cloud Bus中有一些本地配置属性  spring.cloud.bus.\*  （例如： spring.cloud.bus.destination  是使用外部中间件的topic的名字）。 通常情况下，默认值就足够了。

查阅 Spring Cloud Stream  的文档，学习更多关于自定义消息代理的配置。

6、追溯总线事件

总线事件（  RemoteApplicationEvent  的子类）可以通过设置 s pring.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" : "\*:\*\*"

**\}**

这个追溯显示  customers:9000  发送了一个  RefreshRemoteApplicationEvent  ，广播到所有的服务，并被  customers:9000  和 stores:8081  收到（ ack 了） 。

要自己处理 ack 信号，你可以为 AckRemoteApplicationEvent 和 SentApplicationEvent 类型添加一个 @EventListener 添加到应用程序中（并启用跟踪）。 然后你可以利用 TraceRepository 并从那里获取数据。

tips：任何总线应用都可以追溯ack，有时候在一个可以对数据执行更复杂查询的中央服务中执行此操作将非常有用。或者你也可以转发到一个专门的追溯服务中。

7、广播自定义事件

总线可以传输 RemoteApplicationEvent  类型的任何事件，但默认传输格式是JSON，而且 deserializer  需要提前知道那些类型会被使用到。要注册一个新的类型，那么这个类型需要在 org.springframework.cloud.bus.event  的子包中。

要自定义事件名称，你可以在自定义类上使用 @JsonTypeName  ，或者依赖使用类的 简单名称的默认策略。注意生产者和消费者都需要访问这个类的定义。

7.1、在自定义包中注册事件

如果你对于你的自定义事件，不想或者不能使用 org.springframework.cloud.bus.event  子包， 则必须使用 @RemoteApplicationEventScan  指定包以扫描 RemoteApplicationEvent  类型的事件。这样指定的包包含子包。

例如，如果你已经自定义了一个名为FooEvent的事件：

**package** com.acme;

**public** **class** FooEvent **extends** RemoteApplicationEvent \{

...

你可以通过以下方式向 deserializer  注册这个事件：

**package** com.acme;

*@Configuration*

*@RemoteApplicationEventScan*

**public** **class** BusConfiguration \{

...

没有指定值的时候，使用 @RemoteApplicationEventScan  注解的类所在的包会被注册。在这个例子中，会使用 BusConfiguration  的包 com.acme来注册。

你也可以通过 @RemoteApplicationEventScan 注解的 value 、 basePackageClasses   或者   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   的示例都是相同的，因为 RemoteApplicationEventScan 注解显式指定了包   com.acme  会被注册。注意，你可以指定扫描多个基包。

from   [http://cloud.spring.io/spring-cloud-static/spring-cloud-bus/1.3.2.RELEASE/single/spring-cloud-bus.html][http_cloud.spring.io_spring-cloud-static_spring-cloud-bus_1.3.2.RELEASE_single_spring-cloud-bus.html]

[mybroker.com]: http://mybroker.com/
[http_cloud.spring.io_spring-cloud-static_spring-cloud-bus_1.3.2.RELEASE_single_spring-cloud-bus.html]: http://cloud.spring.io/spring-cloud-static/spring-cloud-bus/1.3.2.RELEASE/single/spring-cloud-bus.html