事件监听机制（三）Spring Cloud Bus流程分析

╰+哭是因爲堅強的太久メ 2022-03-15 06:14 420阅读 0赞

## 事件监听机制（三）Spring Cloud Bus流程分析 ##

### 事件监听实现流程 ###

1.  事件对象：  
    继承自`java.util.EventObject`对象，由开发者自行定义实现。
2.  事件源：  
    就是触发事件的源头，不同的事件源会触发不同的事件类型。
3.  事件监听器：  
    事件监听器负责监听事件源发出的事件，事件监听器可以通过实现`java.util.EventListener`这个标识接口.，实现事件监听。

### 流程总结 ###

事件源可以注册事件监听器对象，并可以向事件监听器对象发送事件对象，事件发生后，事件源将事件对象发给已经注册的所有事件监听器，监听器对象随后会根据事件对象内的相应方法响应这个事件。

### 相关实现 ###

#### 相关依赖 ####

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

#### 事件跟踪 ####

# 失效管理端安全 - 
    management.security.enabled = false
    # 默认事件跟踪功能是失效，需要通过配置项激活：
    spring.cloud.bus.trace.enabled=true

#### 访问 http://localhost:8080/beans ####

确认当前 Application Context ID

{ 
        "context": "user-service-client:8080",
        "parent": "user-service-client",
        "beans": []
    }

#### 单点传播 ####

POST http://localhost:8080/bus/refresh?destination=user-service-client:8080

执行 curl：

curl -X POST http://localhost:8080/bus/refresh?destination=user-service-client:8080

日志输出：

INFO 28041 --- [nio-8080-exec-3] o.s.cloud.bus.event.RefreshListener      : Received remote refresh request. Keys refreshed []

#### 集群传播 ####

POST http://localhost:8080/bus/refresh?destination=user-service-client:**

执行 curl：

curl -X POST http://localhost:8080/bus/refresh?destination=user-service-client:**

日志输出：

INFO 28041 --- [nio-8080-exec-5] o.s.cloud.bus.event.RefreshListener      : Received remote refresh request. Keys refreshed []

通过日志可知事件监听器均为：`org.springframework.cloud.bus.event.RefreshListener`

public class RefreshListener
    		implements ApplicationListener<RefreshRemoteApplicationEvent> { 
    
    	private static Log log = LogFactory.getLog(RefreshListener.class);
    
    	private ContextRefresher contextRefresher;
    
    	public RefreshListener(ContextRefresher contextRefresher) { 
    		this.contextRefresher = contextRefresher;
    	}
    
    	@Override
    	public void onApplicationEvent(RefreshRemoteApplicationEvent event) { 
    		Set<String> keys = contextRefresher.refresh();
    		log.info("Received remote refresh request. Keys refreshed " + keys);
    	}
    }

`RefreshListener`监听事件 `RefreshRemoteApplicationEvent`  
自定义 `RefreshRemoteApplicationEvent` 监听器

/** * @ClassName: BusConfiguration * @Description: 自定义监听 * @Author: 尚先生 * @CreateDate: 2019/3/1 9:10 * @Version: 1.0 */
     @Configuration
    public class BusConfiguration { 
    
        @EventListener
        public void onRefreshRemoteApplicationEvent(RefreshRemoteApplicationEvent event) { 
    
            System.out.printf(" Source : %s , originService : %s , destinationService : %s \n",
                    event.getSource(),
                    event.getOriginService(),
                    event.getDestinationService());
    
        }
    }

### 源码分析 ###

分析`org.springframework.cloud.bus.BusAutoConfiguration`

@Configuration
    @ConditionalOnBusEnabled
    @EnableBinding(SpringCloudBusClient.class)
    @EnableConfigurationProperties(BusProperties.class)
    public class BusAutoConfiguration implements ApplicationEventPublisherAware { 
    
    	public static final String BUS_PATH_MATCHER_NAME = "busPathMatcher";
    
    	//绑定消息通道
    	@Autowired
    	@Output(SpringCloudBusClient.OUTPUT)
    	private MessageChannel cloudBusOutboundChannel;
    	// 消息来源校验
    	@Autowired
    	private ServiceMatcher serviceMatcher;
    	// 属性绑定
    	@Autowired
    	private ChannelBindingServiceProperties bindings;
    	// Bus属性
    	@Autowired
    	private BusProperties bus;
    	// 消息发布器
    	private ApplicationEventPublisher applicationEventPublisher;
    
    	// spring加载过程初始化bindings相关属性
    	@PostConstruct
    	public void init() { 
    		BindingProperties inputBinding = this.bindings.getBindings()
    				.get(SpringCloudBusClient.INPUT);
    		if (inputBinding == null) { 
    			this.bindings.getBindings().put(SpringCloudBusClient.INPUT,
    					new BindingProperties());
    		}
    		BindingProperties input = this.bindings.getBindings()
    				.get(SpringCloudBusClient.INPUT);
    		if (input.getDestination() == null) { 
    			input.setDestination(this.bus.getDestination());
    		}
    		BindingProperties outputBinding = this.bindings.getBindings()
    				.get(SpringCloudBusClient.OUTPUT);
    		if (outputBinding == null) { 
    			this.bindings.getBindings().put(SpringCloudBusClient.OUTPUT,
    					new BindingProperties());
    		}
    		BindingProperties output = this.bindings.getBindings()
    				.get(SpringCloudBusClient.OUTPUT);
    		if (output.getDestination() == null) { 
    			output.setDestination(this.bus.getDestination());
    		}
    	}
    
    	@Override
    	public void setApplicationEventPublisher(
    			ApplicationEventPublisher applicationEventPublisher) { 
    		this.applicationEventPublisher = applicationEventPublisher;
    	}
    
    	//监听 Spring Event（本地事件）
    	// 由于@EventListener 监听 Spring Event，事件RemoteApplicationEvent 属于本地事件，因此必然有发布该事件的源头。
    	@EventListener(classes = RemoteApplicationEvent.class)
    	public void acceptLocal(RemoteApplicationEvent event) { 
    		if (this.serviceMatcher.isFromSelf(event)
    				&& !(event instanceof AckRemoteApplicationEvent)) { 
    			this.cloudBusOutboundChannel.send(MessageBuilder.withPayload(event).build());
    		}
    	}
    
    	// 监听 Stream 事件（远程事件）
    	// acceptRemote 监听 Stream 事件，同时发送 Spring Event（本地事件）
    	@StreamListener(SpringCloudBusClient.INPUT)
    	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);
    			}
    			// If it's an ACK we are finished processing at this point
    			return;
    		}
    		// ServiceMatcher#isForSelf(RemoteApplicationEvent) 用于匹配 RemoteApplicationEvent 是否为当前应用实例而来
    		if (this.serviceMatcher.isForSelf(event)
    				&& this.applicationEventPublisher != null) { 
    				// ServiceMatcher#isFromSelf(RemoteApplicationEvent) 用于判断当前事件是否为自己发送
    			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());
    						// 判断事件是为自己发送，执行RefreshRemoteApplicationEvent 事件监听
    						// 如果ACK是激活的，发送AckRemoteApplicationEvent 到管道里
    				this.applicationEventPublisher.publishEvent(ack);
    			}
    		}
    		if (this.bus.getTrace().isEnabled() && this.applicationEventPublisher != null) { 
    			// We are set to register sent events so publish it for local consumption,
    			// irrespective of the origin
    			this.applicationEventPublisher.publishEvent(new SentApplicationEvent(this,
    					event.getOriginService(), event.getDestinationService(),
    					event.getId(), event.getClass()));
    		}
    	}
    	// ...忽略部分源码...
    	// RemoteApplicationEvent自定义实现，可参考后续文章
    }

### 整体流程 ###

假设 user-service-client:8080 执行/bus/refresh 端口，发送了一个RefreshRemoteApplicationEvent事件:

*  curl -X POST http://localhost:8080/bus/refresh?destination=user-service-client:8082
 *  user-service-client:8080 : Bus 事件的发布者、监听者
 *  user-service-client:8081 : Bus 事件监听者
 *  user-service-client:8082: Bus 事件监听者

当 Stream Binder 接收到发布者RefreshRemoteApplicationEvent事件，广播该事件到所有的监听者：

*  user-service-client:8080 : 判断事件是自己发送，SentApplicationEvent
 *  user-service-client:8081：判断事件不是为自己发送，忽略
 *  user-service-client:8082：判断事件是为自己发送，执行RefreshRemoteApplicationEvent 事件监听。如果 ack 激活的，cloudBusOutboundChannel 会发送AckRemoteApplicationEvent 到管道里

### 往期文章 ###

博客地址：https://blog.csdn.net/shang_xs
    微信公众号地址：http://mp.weixin.qq.com/mp/homepage?__biz=MzUxMzk4MDc1OQ==&hid=2&sn=c6c58c06f6f8403af27b6743648b3055&scene=18#wechat_redirect

### 完整代码及详情 ###

具体依赖及相关源码参照

https://github.com/dwyanewede/segmentfault-lessons/tree/master/spring-cloud

### 更多文章 ###

> 事件监听机制（一）Java事件监听  
> https://blog.csdn.net/shang\_xs/article/details/87911756  
> 事件监听机制（二）Spring事件监听  
> https://blog.csdn.net/shang\_xs/article/details/88048545  
> 事件监听机制（三）Spring Cloud Bus流程分析  
> https://blog.csdn.net/shang\_xs/article/details/88050196  
> 事件监听机制（四）从Java事件监听到Spring事件监听  
> https://blog.csdn.net/shang\_xs/article/details/86560994  
> 事件监听机制（五）再话Jdk事件监听到Spring框架事件监听  
> https://blog.csdn.net/shang\_xs/article/details/119794917