Spring Cloud Bus中的事件的订阅与发布(二)

Spring Cloud Bus中的事件的订阅与发布(二)

大家好,又见面了,我是全栈君。

在之前的文章Spring Cloud Bus中的事件的订阅与发布(一)介绍了消息总线的相关事件。 本文主要介绍消息总线的事件监听器以及消息的订阅与发布。

事件监听器

Spring Cloud Bus中,事件监听器的定义可以是实现ApplicationListener接口,或者是使用@EventListener注解的形式。我们看一下事件监听器的类图。

监听器

ApplicationListener接口实现有两个:刷新监听器
RefreshListener和环境变更监听器
EnvironmentChangeListener

RefreshListener

RefreshListener对应的事件是RefreshRemoteApplicationEvent

public class RefreshListener implements ApplicationListener<RefreshRemoteApplicationEvent> {
	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);
	}
}
复制代码

对于刷新时间的处理,调用ContextRefresherrefresh()方法,而定义在Spring Cloud Context中的ContextRefresher用于提供上下文刷新的功能。我们具体看一下refresh()方法。

	public synchronized Set<String> refresh() {
		Map<String, Object> before = extract(
				this.context.getEnvironment().getPropertySources());
		addConfigFilesToEnvironment();
		Set<String> keys = changes(before,
				extract(this.context.getEnvironment().getPropertySources())).keySet();
		this.context.publishEvent(new EnvironmentChangeEvent(keys));
		this.scope.refreshAll();
		return keys;
	}
复制代码

实现很简单,先获取之前环境变量的key-value,然后重新加载新的配置环境文件,通过比对新旧环境变量的map集合,然后发布新的环境变更EnvironmentChangeEvent的事件。this.scope.refreshAll()销毁了在这个范围内,当前实例的所有bean并在下次方法的执行时强制刷新。

EnvironmentChangeListener

EnvironmentChangeListener对应的事件类是EnvironmentChangeRemoteApplicationEvent

public class EnvironmentChangeListener implements ApplicationListener<EnvironmentChangeRemoteApplicationEvent> {
	@Autowired
	private EnvironmentManager env;

	@Override
	public void onApplicationEvent(EnvironmentChangeRemoteApplicationEvent event) {
		Map<String, String> values = event.getValues();
		for (Map.Entry<String, String> entry : values.entrySet()) {
			env.setProperty(entry.getKey(), entry.getValue());
		}
	}
}
复制代码

RefreshListener的实现中,可以知道该事件的实现最终又发布了一个新的事件EnvironmentChangeListener。在刷新监听器中,构造了变更了的环境变量的map,交给环境变更监听器。上面对环境变更事件的处理,遍历变更了的配置环境属性,并在本地应用程序的环境中将新的属性值设置到对应的键。

TraceListener

TraceListener的实现是通过注解@EventListener的形式,监听的事件为:确认事件AckRemoteApplicationEvent和发送事件SentApplicationEvent

@EventListener
	public void onAck(AckRemoteApplicationEvent event) {
		this.repository.add(getReceivedTrace(event));
	}

	@EventListener
	public void onSend(SentApplicationEvent event) {
		this.repository.add(getSentTrace(event));
	}

	protected Map<String, Object> getSentTrace(SentApplicationEvent event) {
		Map<String, Object> map = new LinkedHashMap<String, Object>();
		map.put("signal", "spring.cloud.bus.sent");
		map.put("type", event.getType().getSimpleName());
		map.put("id", event.getId());
		map.put("origin", event.getOriginService());
		map.put("destination", event.getDestinationService());
		if (log.isDebugEnabled()) {
			log.debug(map);
		}
		return map;
	}

	protected Map<String, Object> getReceivedTrace(AckRemoteApplicationEvent event) {
		Map<String, Object> map = new LinkedHashMap<String, Object>();
		map.put("signal", "spring.cloud.bus.ack");
		map.put("event", event.getEvent().getSimpleName());
		map.put("id", event.getAckId());
		map.put("origin", event.getOriginService());
		map.put("destination", event.getAckDestinationService());
		if (log.isDebugEnabled()) {
			log.debug(map);
		}
		return map;
	}
复制代码

在SentTrace中,主要记录了signal、事件类型type、id、源服务origin和目的服务destination的属性值。而在ReceivedTrace中,表示对事件的确认,主要记录了signal、事件类型event、id、源服务origin和目的服务destination的属性值。这些信息默认存储于内存中,可以通过/trace端点获取最近的事件信息,如下图所示:

{
    "timestamp": 1517229555629,
    "info": {
        "signal": "spring.cloud.bus.sent",
        "type": "RefreshRemoteApplicationEvent",
        "id": "c73a9792-9409-47af-993c-65526edf0070",
        "origin": "config-server:8888",
        "destination": "config-client:8000:**"
    }
},
{
	"timestamp": 1517227659384,
	"info": {
	    "signal": "spring.cloud.bus.ack",
	    "event": "RefreshRemoteApplicationEvent",
	    "id": "846f3a17-c344-4d29-93f3-01b73c5bf58f",
	    "origin": "config-client:8000",
	    "destination": "config-client:8000:**"
	}
}
复制代码

至于事件的发起,我们将在下一节结合消息的订阅与发布一起讲解。

消息的订阅与发布

Spring Cloud Bus基于Spring Cloud Stream,对特定主题的消息进行订阅与发布,事件以消息的形式传递到其他服务实例。

通道定义

既然是基于stream,我们首先看一下input和output的通道定义。

public interface SpringCloudBusClient {

String INPUT = "springCloudBusInput";

String OUTPUT = "springCloudBusOutput";

@Output(SpringCloudBusClient.OUTPUT)
MessageChannel springCloudBusOutput();

@Input(SpringCloudBusClient.INPUT)
SubscribableChannel springCloudBusInput();
}
复制代码

可以看到,bus中定义了springCloudBusInputspringCloudBusOutput两个通道,分别用于定于订阅与发布springCloudBus的消息。

bus属性定义

其次,我们看一下bus中关于stream的属性定义。在基础应用中我们就知道bus订阅的话题是springCloudBus,下面看一下在bus中的其他属性的定义。

@ConfigurationProperties("spring.cloud.bus")
public class BusProperties {

//环境变更相关的属性
private Env env = new Env();
// 刷新事件相关的属性
private Refresh refresh = new Refresh();
//与ack相关的属性
private Ack ack = new Ack();
//与追踪ack相关的属性
private Trace trace = new Trace();
//Spring Cloud Stream消息的话题
private String destination = "springCloudBus";

//标志位,bus是否可用
private boolean enabled = true;

...
}
复制代码

上面的bus属性,设置了一些默认值,正好与事实也是相符的,我们没有进行任何spring.cloud.bus配置也能够进行正常运行。通过在配置文件中修改相应的属性,实现bus的更多功能扩展。env、refresh、ack和trace分别对应不同的事件,在配置文件中有一个开关属性,默认都是开启的,我们可以根据需要进行关闭。

消息的监听与发送

上面两部分讲了stream通道和基本属性的定义,最后我们看下bus中对指定主题的消息如何发送与监听处理。在META-INF/spring.factories配置了EnableAutoConfiguration配置项为BusAutoConfiguration,在服务启动时会自动加载到Spring容器中,其中对于指定主题的消息如何发送与监听处理如下:

@Configuration
@ConditionalOnBusEnabled //bus启用的开关
@EnableBinding(SpringCloudBusClient.class) //绑定通道
@EnableConfigurationProperties(BusProperties.class)
public class BusAutoConfiguration implements ApplicationEventPublisherAware {

//注入source接口,用于发送消息
@Autowired
@Output(SpringCloudBusClient.OUTPUT)
private MessageChannel cloudBusOutboundChannel;

// 监听RemoteApplicationEvent事件
@EventListener(classes = RemoteApplicationEvent.class)
public void acceptLocal(RemoteApplicationEvent event) {
    if (this.serviceMatcher.isFromSelf(event)
            && !(event instanceof AckRemoteApplicationEvent)) {
        //当事件是来自自己的并且不是ack事件,则发送消息
    this.cloudBusOutboundChannel.send(MessageBuilder.withPayload(event).build());
    }
}
//消息的消费,也是事件的发起
@StreamListener(SpringCloudBusClient.INPUT)
public void acceptRemote(RemoteApplicationEvent event) {
    if (event instanceof AckRemoteApplicationEvent) {
        //ack事件
        if (this.bus.getTrace().isEnabled() && !this.serviceMatcher.isFromSelf(event)
                && this.applicationEventPublisher != null) {
            //当开启bus追踪且不是自己的ack事件,则通知所有的注册该事件的监听者,否则直接返回
            this.applicationEventPublisher.publishEvent(event);
        }
        return;
    }
    //消费消息,该消息属于自己
    if (this.serviceMatcher.isForSelf(event)
            && this.applicationEventPublisher != null) {
        //不是自己发布的事件,正常处理
        if (!this.serviceMatcher.isFromSelf(event)) {
            this.applicationEventPublisher.publishEvent(event);
        }
        //消费之后,需要发送ack确认事件
        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()));
    }
}

//...
}
复制代码

@ConditionalOnBusEnabled注解是bus的开关,默认开启。@EnableBinding绑定了SpringCloudBusClient中定义的通道。在应用服务启动时,自动化配置类加载了bus的API端点、刷新、ACK追踪以及bus环境变量的配置等beans。 @Output表示输出output绑定目标将由框架创建,由该通道发送消息。 还涉及到上面列出来的两个主要方法:acceptLocalacceptRemote

acceptLocal是一个基于注解实现的事件监听器,监听的事件类型是RemoteApplicationEvent,对于该事件的处理方法是,当事件是来自自己的并且不是ack事件,则发送消息。

@StreamListener注解是Spring Cloud Stream中提供的,用来标识一个方法作为@EnableBinding绑定的input通道的监听器。acceptRemote方法, 传递的参数RemoteApplicationEvent就是stream中的消息。如果是确认类事件,当开启了事件追踪且事件不是来自于自身,则发布该事件,对于确认类事件,处理已经完成; 如果自身需要处理该事件且该事件不是来自自身,则发布该事件。需要注意的是,当开启事件追踪时,构造一个确认事件并将该事件发布;最后,当开启了事件追踪,这边的处理是注册已发送的事件,以便发布供本地消费,而不论其来源。

总结

本文在上一篇介绍Spring Cloud Bus中的事件基础上,结合源码继续介绍事件的监听器以及事件的订阅与发布是如何在消息总线中实现的。 消息总线常用于传播状态的变更和管理指令的发布。而消息总线最常用的场景就是更新应用服务的配置信息,需要结合Config Server使用,当然消息总线的实现其实是基于Spring Cloud Stream,Stream封装了各种不同的MQ中间件,产生的消息实则是推送配置信息的变更。

订阅最新文章,欢迎关注我的公众号

参考

Spring Cloud Bus-v1.3.3

版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。

发布者:全栈程序员-用户IM,转载请注明出处:https://javaforall.cn/107670.html原文链接:https://javaforall.cn

【正版授权,激活自己账号】: Jetbrains全家桶Ide使用,1年售后保障,每天仅需1毛

【官方授权 正版激活】: 官方授权 正版激活 支持Jetbrains家族下所有IDE 使用个人JB账号...

(0)


相关推荐

  • js基本七种数据类型_js原始数据类型

    js基本七种数据类型_js原始数据类型Js中的数据类型虽然也是一个老生常谈的问题,但它经常出现在整个面试的前几问中,面试官会通过你的回答来决定之后问题的走向,比如当你回答基本数据类型时少回答了一个String时,那么面试官很可能就会问你String都有哪写方法哦~

    2022年10月26日
  • HTTP常见的状态码

    HTTP常见的状态码状态码解释200一切正常301永久重定向302临时重定向401用户名或密码错误403禁止访问(客户端IP地址被拒绝404文件不存在414请求URI头部过长500服务器内部错误502BadGateway(错误网关)…

  • VS导入没有.sln文件的项目[通俗易懂]

    VS导入没有.sln文件的项目[通俗易懂]关于在VS中导入没有.sln文件的.net项目     导入的是一个网站项目,步骤是先创建一个空白的解决方案,再在解决方案中点击右键,添加一个现有网站项目,再选择相应的项目文件,即可进行网站的运行。如果在运行过程中出现xxx.cs文件中不存在该类型的命名或命名空间,那么极有可能是相应的.net框架版本不一致导致,所以在点击该项目的属性页中调整相应的框架版本,即可进行相应的网站项目运行。…

  • windows批处理脚本_批处理打开文件命令

    windows批处理脚本_批处理打开文件命令 Windows平台批处理命令教程 do2jiang@gmail.com蒋冬冬 收集于网络 2009.4  该教程一共分为4大部分,第一部分是批处理的专用命令,第二部分是特殊的符号与批处理,第三部分是批处理与变量,第四部分是完整案例。 第一部分:批处理的专用命令 批处理文件是将一系列命令按一定的顺序集合为一个可执行的文本文件,其扩

  • python中sqrt的用法_Python中sqrt函数怎么用「建议收藏」

    python中sqrt的用法_Python中sqrt函数怎么用「建议收藏」Python中sqrt函数怎么用?下面给大家带来sqrt函数的相关介绍:Python数字sqrt()函数返回x的平方根(x>0)。语法以下是sqrt()方法的语法-importmathmath.sqrt(x)注意-此函数不可直接访问,需要导入math模块,然后需要使用math静态对象调用此函数。参数x-这是一个数字表达式。返回值该方法返回x的平方根(x>0)。sqrt()…

  • 从单片机到ARM Linux驱动——Linux驱动入门篇

    从单片机到ARM Linux驱动——Linux驱动入门篇大一到大二这段时间里学习过单片机的相关知识,对单片机有一定的认识和了解。如果要深究其原理可能还差了一些火候。知道如何编写程序来点量一个LED灯,改一改官方提供的例程来实现一些功能做一些小东西,对IIC、SPI底层的通信协议有一定的了解,但是学着学着逐渐觉得单片机我也就只能改改代码了(当然有的代码也不一定能改出来)。对于我这种以后只想搬砖的码农来说已经差不多了(仅仅是个人观点),…

发表回复

您的电子邮箱地址不会被公开。

关注全栈程序员社区公众号