SpringCloud bus 使用介绍

对于springcloud bus 原理的介绍，主要是springcloud streaem 配合事件监听

1. spring cloud bus介绍

Spring cloud bus使用轻量级消息代理将分布式系统的节点连接起来，可以使用此代理，广播状态更改(例如配置更改)或其他管理指令。它可以用作应用程序之间的通信通道。该项目提供了两种消息传输处理：AMQP broker 和Kafka 2. 如果你想使用activemq或其他的消息中间件作为消息传输，那么需要实现spring cloud stream消息驱动的绑定（spring cloud bus其实也是使用stream实现的rabbitmq和kafka）

2. 构建项目

我们使用rabbitmq作为中间件

引入依赖：

1
2
3
4
<dependency>    
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-bus-amqp</artifactId>
</dependency>

application.yml中添加连接rabbitmq的配置：

1
2
3
4
5
6
spring:  
    rabbitmq:    
        host: mybroker.com    
        port: 5672    
        username: user    
        password: secret

3. 端点

spring cloud bus 提供了两个端点/actuator/bus-refresh and /actuator/bus-env，他们分别对应了Spring Cloud Commons中的/actuator/refresh和/actuator/env

 

3.1 /actuator/bus-refresh

可以清除RefreshScope中的缓存并且重新绑定@ConfigurationProperties，一般是配合config组件使用，当config server 中的配置修改时可动态刷新config client中加载的yml配置

暴露端口：

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

 

3.2 /actuator/bus-env

更新每个实例中指定的environment 键值对

暴露端口：

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

3.3 destination参数

可以为端口后添加destination参数来指定你想广播的服务实例。

假设我们现在使用/bus-refresh端点

当你想指定某个具体服务实例更新时：/actuator/bus-refresh?destination=customers:9000

   参数值为(spring.application.name:server.port)

当你想指定服务更新时：/actuator/bus-refresh?destination=customers:**

   **表示customers服务的所有实例

4. 事件追踪

bus中的事件大致只有3中：

  • EnvironmentChangeRemoteApplicationEvent：对应/bus-env的端点事件
  • RefreshRemoteApplicationEvent：对应/bus-refresh端点的事件
  • AckRemoteApplicationEvent：这是一个确认事件，没有什么含义，只是说确认事件已发送或已接受

 

1
spring.cloud.bus.trace.enabled=true

默认是关闭的，因为它是靠本地项目的内存来存储追踪记录的，会比较消耗资源。

开启后会显示发送的每个事件和来自每个服务实例的所有确认事件（ack）

通过访问端点/actuator/trace可查看追踪记录

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
{
	"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服务实例发出(signal=…sent),并广播了所有的服务实例。

轨迹：我们这里是查看的customers:9000该实例的追踪记录，当请求bus-refresh端口后，该实例显示确认了该事件（第3个json），然后发送了事件到消息中间件中（第2个json），最后因为接受到了来自消息中间件的远程事件，所以再次打印了ack确认日志（第一个json）

5. 源码分析

我们分析下/bus-refresh的整个操作流程

5.1 Spring本地事件发布

需查看类：RefreshBusEndpoint

因为我们是通过/bus-refresh端点作为入口更新系统的配置信息的，所以我们可以找到RefreshBusEndpoint类：

1
2
3
4
5
6
7
8
@Endpoint(id = "bus-refresh")
public class RefreshBusEndpoint extends AbstractBusEndpoint {     

    @WriteOperation
    public void busRefresh() {
        this.publish(new RefreshRemoteApplicationEvent(this, this.getInstanceId(), (String) null));
    }
}

非总线刷新也有一个刷新配置接口，

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
@ConfigurationProperties(prefix = "endpoints.refresh", ignoreUnknownFields = false)
@ManagedResource
public class RefreshEndpoint extends AbstractEndpoint<Collection<String>> {

	private ContextRefresher contextRefresher;

	public RefreshEndpoint(ContextRefresher contextRefresher) {
		super("refresh");
		this.contextRefresher = contextRefresher;
	}

	@ManagedOperation
	public String[] refresh() {
		Set<String> keys = contextRefresher.refresh();
		return keys.toArray(new String[keys.size()]);
	}

	@Override
	public Collection<String> invoke() {
		return Arrays.asList(refresh());
	}

}

通过该类，我们发现，其实在我们调用/bus-refresh端口时，它是发布了一个RefreshRemoteApplicationEvent事件（spring本地事件），注意是本地事件。RefreshRemoteApplicationEvent的父类是RemoteApplicationEvent

5.2 通过Spring本地事件监听，将消息发送到了消息中间件中

查看类：BusAutoConfiguration

在BusAutoConfiguration配置类中，我们找到了这么一段代码：

1
2
3
4
5
6
7

@EventListener(classes = {RemoteApplicationEvent.class})
public void acceptLocal(RemoteApplicationEvent event) {
    if (this.serviceMatcher.isFromSelf(event) && !(event instanceof AckRemoteApplicationEvent)) {
        this.cloudBusOutboundChannel.send(MessageBuilder.withPayload(event).build());
    }
}

发现在这个配置类中，有个本地事件监听器，且监听的是RemoteApplicationEvent事件，而在RefreshBusEndpoint中发布的事件对象的父类正好就是RemoteApplicationEvent事件，所以我们可以断定，这个监听器是能够监听到刚刚发布的事件的。

 从代码中可以看出来它进行了判断，从方法名上就可以看出来，判断了这个监听到的事件是否是自己发的，并且是否是Ack事件，当然我们这里是满足的，所以执行了this.cloudBusOutboundChannel.send()方法，这个方法的意思就是向管道中发送消息，这个管道其实就是spring cloud stream 与 消息中间件绑定的管道，通俗来讲就是它把这个事件发送到rabbitmq上去了。

 

补充：this.cloudBusOutboundChannel对象其实是和消息中间件的channel通道相连接的，查找BusAutoConfiguration类发现，其实该对象和rabbitmq相连接的exchange名称是springCloudBus

1
2
3
4
5
6
 @Autowired
@Output("springCloudBusOutput")
public void setCloudBusOutboundChannel(MessageChannel cloudBusOutboundChannel) {
    this.cloudBusOutboundChannel = cloudBusOutboundChannel;
}

5.3 Stream远程事件监听

查看类：BusAutoConfiguration

通过上面的操作，事件已经被发送到rabbitmq上去了，那我们本地就要进行远程事件的监听了，监听的就是this.cloudBusOutboundChannel对象和rabbitmq相连接的通道,远程事件的监听使用 @StreamListener()注解

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
@StreamListener("springCloudBusInput")
public void acceptRemote(RemoteApplicationEvent event) {
    if (event instanceof AckRemoteApplicationEvent) {
        if (this.bus.getTrace().isEnabled() && !this.serviceMatcher.isFromSelf(event) && this.applicationEventPublisher != null) {
            this.applicationEventPublisher.publishEvent(event);
        }
    } else {
        if (this.serviceMatcher.isForSelf(event) && this.applicationEventPublisher != null) {
            if (!this.serviceMatcher.isFromSelf(event)) {
                this.applicationEventPublisher.publishEvent(event);
            }
            if (this.bus.getAck().isEnabled()) {
                AckRemoteApplicationEvent ack = new AckRemoteApplicationEvent(this, this.serviceMatcher.getServiceId(), this.bus.getAck().getDestinationService(), event.getDestinationService(), event.getId(), event.getClass());
                this.cloudBusOutboundChannel.send(MessageBuilder.withPayload(ack).build());
                this.applicationEventPublisher.publishEvent(ack);
            }
        }
        if (this.bus.getTrace().isEnabled() && this.applicationEventPublisher != null) {
            this.applicationEventPublisher.publishEvent(new SentApplicationEvent(this, event.getOriginService(), event.getDestinationService(), event.getId(), event.getClass()));
        }
    }
}
1
查看代码我们可以发现接受到事件后会经过很多判断，判断是否是ack事件，或该事件是不是自己发送的(this.serviceMatcher.isFromSelf(event)，或者该事件是否为我而来，即消息是不是发给我的(his.serviceMatcher.isForSelf(event)：上面讲过当/bus-refresh时可以指定服务实例发送事件，所以虽然能监听到事件，但不一定这个事件就是针对你的

截取下代码：

1
2
3
4
5
6
if(this.serviceMatcher.isForSelf(event)&&this.applicationEventPublisher!=null){
    if (!this.serviceMatcher.isFromSelf(event)) {
        this.applicationEventPublisher.publishEvent(event);
    }
}

1
这段代码意思是：事件就是针对我来的且不是我自己发送的这个事件，那么就又会在本地发送该事件，发送后会被5.4节中的RefreshListener所监听到。如果是自己发的，那么就不做任何处理了，因为他在发送消息到中间件时，就已经被RefreshListener监听器处理过了。

5.4 Spring本地事件监听

查看：RefreshListener类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
public class RefreshListener implements ApplicationListener<RefreshRemoteApplicationEvent> {
    private static Log log = LogFactory.getLog(RefreshListener.class);
    private ContextRefresher contextRefresher;

    public RefreshListener(ContextRefresher contextRefresher) {
        this.contextRefresher = contextRefresher;
    }

    public void onApplicationEvent(RefreshRemoteApplicationEvent event) {
        Set<String> keys = this.contextRefresher.refresh();
        log.info("Received remote refresh request. Keys refreshed " + keys);
    }
}

该类是通过实现ApplicationListener的接口来监听RefreshRemoteApplicationEvent事件的，然后执行了this.contextRefresher.refresh()方法来刷新程序中的context上下文。

5.5 流程总结

1. 假设我们访问端口为8080的user服务，当访问端点/bus-refresh时，会被RefreshBusEndpoint执行，该类会发送一个spring本地事件，事件名为RefreshRemoteApplicationEvent（在8080这个实例中进行事件传播）

2. 发送后，会被两个位置的监听器监听：

  • 一个是RefreshListener，该监听器监听到后直接refresh上下文
  • 一个是BusAutoConfiguration，该配置类中存在监听RemoteApplicationEvent事件的监听器，被这个监听器监听后，会向rabbitmq发送事件消息实体

3. 在bus中的应用都会进行strem的远程事件监听（@StreamListener(“springCloudBusInput”)，应用监听到strem事件后，会判断事件的发送和发往的情况

  • 如果该事件是发向自己的，并且不是自己发出去的，那么会再次将这个event通过spring本地事件发出去，让当前实例的RefreshListener监听器监听到并执行refresh操作
  • 如果该事件既是自己发出去的，也是发向自己的，那么就不执行任何操作了，因为在发出去的时候已经被RefreshListener监听器监听过了
  • 如果事件不是发向自己的，不执行任何操作

6. 通过bus实现自定义事件发送

6.1 创建事件

事件需要继承RemoteApplicationEvent类

1
2
3
4
5
6
7
8
9
public class CustomEvent extends RemoteApplicationEvent {
    private CustomEvent() {
        //一定要有，序列化时会用到 
    }

    public CustomEvent(String msg, String originService, String destinationService) {
        super(msg, originService, destinationService);
    }
}

6.2 添加配置类

1
2
3
4
@Configuration
@RemoteApplicationEventScan(basePackageClasses = CustomEvent.class)
public class BusConfiguration {
}
1
@RemoteApplicationEventScan(basePackageClasses = CustomEvent.class)用来告诉bus自己实现的事件在哪个包下，也可直接写在启动类上

6.3  添加事件监听器

1
2
3
4
5
6
7
8
9
10
@Configuration
public class CustomEventListener {
    @Value("${server.port}")
    private String port;

    @EventListener
    public void onCustomRemoteApplicationEvent(CustomEvent event) {
        System.out.printf("CustomRemoteApplicationEvent - " + " port : %s , Source : %s , originService : %s , destinationService : %s \n", port, event.getSource(), event.getOriginService(), event.getDestinationService());
    }
}

我们这里监听器只是打印了一句话

6.4 事件发布者

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
@RestController
public class Controller {
    /**
     * Spring Cloud bus 外部化配置
     */
    @Autowired
    private BusProperties busProperties;
    /**
     * 事件发布者
     */
    @Autowired
    private ApplicationEventPublisher eventPublisher;

    @PostMapping("/bus/event/publish/custom")
    public boolean publishUserEvent(String msg, @RequestParam(value = "destination", required = false) String destination) {
        String instanceId = busProperties.getId();
        CustomEvent event = new CustomEvent(msg, instanceId, destination);
        eventPublisher.publishEvent(event);
        return true;
    }
}

代码中的instanceId可以通过spring.cloud.bus.id属性设置：

下面是官网的原话，我谷歌翻译了以下

6.5 配置文件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
server:  
    port: 8082
management:  
    endpoints:    
        web:      
            exposure:        
                include: "*"
spring:  
    rabbitmq:    
        host: 127.0.0.1    
        port: 5672    
        username: springcloud    
        password: springcloud  
    application:    
        name: bus-custom-event

6.5启动测试

分别以8082和8083端口启动项目（ideal中可以通过修改program arguments参数来对同一个项目并行启动）

启动后，访问：http://localhost:8082/bus/event/publish/custom?destination=bus-custom-event:8083

会发现启动的两个项目都可以监听到事件并打印信息

Java
#spring #springcloud
SpringCloud bus 使用介绍
https://kanchai.club/2022/12/10/学习spring cloud bus 使用，自定义BUS事件/
作者
625
发布于
2022年12月10日
许可协议