Spring Cloud 基于Bus 的AB-TEST组件

时间  2019-11-17
标签 spring cloud 基于 bus test 组件 栏目 Spring 繁體版
原文   原文链接

1、前情提要：

因剧情须要，因此准备在基础开发平台中进行AB-TEST组件开发。目前主要使用了Spring Cloud E-SR2 版本，其中使用了kafka做为内置bus总线，另外一组kafka用于监控trace推送（如zipkin、自定义监控）。AB-TEST你们都应该了解过,如不了解请参考www.itcodemonkey.com/article/103… ，这里就很少讲了。 其实简单来讲就是根据配置好的分桶模式如A 、B、C ，在进行逻辑处理时根据某一类条件（如uid）计算出当前用户分桶进行动态的逻辑处理（简单来讲就是多个if else）。html

2、方案选型：

如要作成组件化那必然要对上层开发透明，最好时无感知的进行逻辑切换，固然咱们第一步不须要那么完美，先实现功能组件。在进行技术选型的时候参考了两种模式：
一、zookeeper
优势：技术简单，有定义好的工具
缺点：增长应用依赖的组件，当zk出现问题时形成ab-test失效
二、bus总线
优势：实现简单，耦合度低
缺点：须要详解cloud bus 机制
固然咱们选择了后者，由于在基础平台中，组件的侵入性越低，对上层开发越友好，并且就越稳定。spring

3、Spring CLoud Bus 事件机制

由于目前使用的是Spring Cloud E SR2全家桶，因此在技术处理上也遵循其原则，尽可能使用内置技术栈实现。内部主要以cloud bus机制进行了简单的扩展实现，下面咱们先来简单了解下BUS EVENT机制。bash

Bus 机制主要创建在 stream的基础之上，在cloud的下 咱们能够选择rabbitmq 或者kafka，其特色主要是针对spring event消息的订阅与发布。app

Spring Event 事件驱动模型 dom

image.png
能够看出模型由三部分构成：

  • 事件：ApplicationEvent，继承自JDK的EventObject，全部事件将继承它，并经过source获得事件源
  • 发布者：ApplicationEventPublisher及ApplicationEventMulticaster接口，使用这个接口，咱们的Service就拥有了发布事件的能力。
  • 订阅者：在spring bus 中能够实现 ApplicationListener（继承自JDK的EventListener），全部监听器将继承它或添加@EventListener

Bus 事件ide

众所周知，在bus使用中，最多的场景就是配置中心的动态配置刷新。也就是说咱们经过/bus/refresh 接口调用就能够进行针对性的配置刷新了，根据这条线，咱们来看下内部的源码结构。工具

一、经过rest 接口进行外部请求 此处cloud 借助其端点监控机制实现，主要看 RefreshBusEndpoint，固然Cloud E 和 F版本有些许不同，但其实不难理解Cloud E SR2组件化

@ManagedResource
public class RefreshBusEndpoint extends AbstractBusEndpoint {

	public RefreshBusEndpoint(ApplicationEventPublisher context, String id,
			BusEndpoint delegate) {
		super(context, id, delegate);
	}
    //定义对外访问接口
	@RequestMapping(value = "refresh", method = RequestMethod.POST)
	@ResponseBody
	@ManagedOperation
	public void refresh(
			@RequestParam(value = "destination", required = false) String destination) {
		publish(new RefreshRemoteApplicationEvent(this, getInstanceId(), destination));
	}

}
复制代码

Cloud F SR2测试

@Endpoint(id = "bus-refresh") //TODO: document new id
public class RefreshBusEndpoint extends AbstractBusEndpoint {

	public RefreshBusEndpoint(ApplicationEventPublisher context, String id) {
		super(context, id);
	}

	@WriteOperation
	public void busRefreshWithDestination(@Selector String destination) { //TODO: document destination
		publish(new RefreshRemoteApplicationEvent(this, getInstanceId(), destination));
	}

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

}
复制代码

经过上面的代码能够看到，请求进来的话都调用了 AbstractBusEndpoint 的publish进行了事件发布优化

public class AbstractBusEndpoint {

	private ApplicationEventPublisher context;

	private String appId;

	public AbstractBusEndpoint(ApplicationEventPublisher context, String appId) {
		this.context = context;
		this.appId = appId;
	}

	protected String getInstanceId() {
		return this.appId;
	}
    //发布事件
	protected void publish(ApplicationEvent event) {
		context.publishEvent(event);
	}

}
复制代码

其中ApplicationEventPublisher 在哪定义的呢，这就不得不说BusAutoConfiguration 这里了，这是bus的核心加载器,在经过外部接口调用发布事件后内部对事件进行了监听和处理就在BusAutoConfiguration中，以下：

//消费事件，进行逻辑判断是不是由本身发出的事件，若是是本身内部发出的事件则经过stream（kafka）进行发布
@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（kafka）进行外部订阅类型为RemoteApplicationEvent 的事件
@StreamListener(SpringCloudBusClient.INPUT)
	public void acceptRemote(RemoteApplicationEvent event) {
        //判断是不是ack类型的回执事件，是的话进行内部发布，用于bustrace 等处理
		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; } //判断是不是给本身的事件，在外部接口请求时可增长destination 进行选择*:*表明所有应用 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); } } //判断是否要进行trace跟踪，默认关闭 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())); } } 复制代码

观看完内部事件消费，和stream消息订阅，那bus 的stream又是怎么进行初始化和工做的呢，答案依然在BusAutoConfiguration 中，以下：

@Autowired
	private BusProperties bus;

	private ApplicationEventPublisher applicationEventPublisher;

	@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());
		}
	}
复制代码

对于cloud stream就很少说了，这里很简单的进行了初始化，因此对于发布和订阅消息就很清晰了。其实在BusAutoConfiguration中因此的事件消费、发布、订阅都是为了集群内部的通讯，而真正的事件处理却不此处。那么对于配置的刷新行为到底在哪呢，通过查看对于刷新的操做要看RefreshListener 了。 以下图 RefreshListener 针对事件进行了监听，其事件使用的是RefreshRemoteApplicationEvent，其继承RemoteApplicationEvent。

image.png 

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);
	}
}
复制代码

经过源码能够看出，事件的刷新行为 Set keys = contextRefresher.refresh(); （固然contextRefresher 定义也是在BusAutoConfiguration 有兴趣能够查看下），对于刷新到底怎么实现的，也就接近了bus config刷新的核心：ContextRefresher

public synchronized Set<String> refresh() {
        //加载当前内存中的配置
		Map<String, Object> before = extract(
				this.context.getEnvironment().getPropertySources());
        //单独启动一个内部容器进行远程配置加载，
		addConfigFilesToEnvironment();
        //交叉对比配置信息，其实就俩Map进行，若是新配置不包含这个key了就result.put(key, null);，若是俩val不同，就            覆盖新值
		Set<String> keys = changes(before,
				extract(this.context.getEnvironment().getPropertySources())).keySet();
        //发布内部变动事件
		this.context.publishEvent(new EnvironmentChangeEvent(context, keys));
        //刷新配置，主要是加了注解@Refresh的方法 参数等，这里的坑在于会形成服务状态变动 UP-》DOWN-》UP 
        //大面积波动的话很可怕
		this.scope.refreshAll();
		return keys;
	}
复制代码

通过以上的源码，脉络就很清晰了： 一、外部POST /bus/refresh 进行了刷新行为，发布内部事件RefreshRemoteApplicationEvent 二、经过@EventListener 进行内部事件消费，若是是本身内部发布的事件就经过stream进行广播 三、经过@StreamListener 对stream进行监听，若是是给本身的事件，就进行内部转发，具体需不须要ack trace则根据配置进行。 四、内部经过RefreshListener 消费事件，经过ContextRefresher.refresh 进行配置刷新 这下一个完整的bus机制就展示在咱们眼前，咱们只须要简单的进行改造，就能实现本身的动态事件推送了。

4、AB-TEST机制实现：

通过上面bus event的铺垫，如今咱们来讲下实现AB-TEST中分为了几个目标阶段：
复制代码

一、应用启动load分桶数据 看过了bus的模式，这里就简单咯。在这里咱们经过继承 PropertyResourceConfigurer 来实现配置的初始化，而配置的来源就是cloud 配置中心的 application.properties，由于使用了配置中心后此配置会自动加载不要额外的处理。而后在对加载的配置进行归类便可（因我为test配置定义了前缀，因此只需过滤其便可），模仿bus配置筛选便可。

public static Map<String, Object> extract(MutablePropertySources propertySources) {
        Map<String, Object> result = new HashMap<>(16);
        List<PropertySource<?>> sources = new ArrayList<PropertySource<?>>();
        for (PropertySource<?> source : propertySources) {
            sources.add(0, source);
        }
        for (PropertySource<?> source : sources) {
            if (!standardSources.contains(source.getName())) {
                extract(source, result);
            }
        }
        return result;
    }

    public static void extract(PropertySource<?> parent, Map<String, Object> result) {
        if (parent instanceof CompositePropertySource) {
            try {
                List<PropertySource<?>> sources = new ArrayList<PropertySource<?>>();
                for (PropertySource<?> source : ((CompositePropertySource) parent)
                        .getPropertySources()) {
                    sources.add(0, source);
                }
                for (PropertySource<?> source : sources) {
                    extract(source, result);
                }
            } catch (Exception e) {
                return;
            }
        } else if (parent instanceof EnumerablePropertySource) {
            for (String key : ((EnumerablePropertySource<?>) parent).getPropertyNames()) {
                result.put(key, parent.getProperty(key));
                log.debug("PropertyConfigure load K[{}] V[{}]", key, parent.getProperty(key));
            }
        }
    }
复制代码

二、在请求来临时进行动态计算分桶 定义自定义注解@NoveTest用于标注须要进行测试的方法，定义NoveParamInterceptor 对入参进行解析，定义NoveTestInterceptor内部拦截器进行注解切入 ，为增长了@NoveTest进行动态分桶计算。

@Pointcut("@annotation(NoveTest)")
    public void anyMethod() {

    } 
@Before(value = "anyMethod()")
    public void doBefore(JoinPoint jp) throws Exception {

        try {
            MethodSignature methodSig = (MethodSignature) jp.getSignature();
            Annotation[] annotations = methodSig.getMethod().getDeclaredAnnotations();
            for (Annotation annotation : annotations) {
                String name = annotation.annotationType().getName();
                String novetest = NoveTest.class.getName();
                if (novetest.equals(name)) {
                    NoveTest test = (NoveTest) annotation;
                    Map<String, String> buckets = RandomBucketUtils.getBuckets(test.name());
                    RandomBucketUtils.loadBuckets(buckets);
                }
            }
        } catch (Exception e) { //防护性容错

        }

    }
复制代码

其中分桶计算的策略很简单，经过uid + 因子进行hash 计算index 获取数据

int hash = (((sole + factor).hashCode() % 100) + 100) % 100;
 //获取参数百分比值
      config.entrySet().stream().sorted(Map.Entry.comparingByKey(Comparator.reverseOrder())).forEach(entry -> {
                if (entry.getKey().contains(percent)) {
                    IntStream.range(0, Integer.valueOf((String) entry.getValue())).forEach(i -> bucket.add(entry.getKey()));
                }

            });
复制代码

好了，定义完成，上层应用开发使用方式：

@RequestMapping("/test")
    @NoveTest(name = {"312", "feed"})
    public Map<String, String> test(@RequestParam(defaultValue = "", required = false) String ma) {

        String type = RandomBucketUtils.getProperties("type");
        //TODO 经过type 分枝处理 if else
        
         return null;

    }
复制代码

三、修改配置进行分桶策略的动态刷新 分桶策略不可能一直不变，并且变化的时候也不该该从新发版，那真是太恶心人了。因此动态配置推送就相当重要了，固然你们也可使用自带的bus总线进行数据推送，可是其destroy的问题真是恶心到家了，并且有可能形成服务大面积瘫痪，慎用。 基于种种问题，就须要自定义bus event，结合其现有的bus进行数据推送。

其实很简单，分为几步：

  • 自定义端点，进行自定义事件发送。（事件广播不须要特殊处理）
  • 自定义Listener进行本地事件消费，进行数据刷新
  • 注意几点，在Cloud E 和F 版本有一些不一样之处。主要在于端点的暴露的策略上，在F版本上使用以下
@WebEndpoint(id = "nove-refresh")
public class NoveRefreshBusEndpoint extends AbstractBusEndpoint {

    public NoveRefreshBusEndpoint(ApplicationEventPublisher context,  String id) {
        super(context, id);
    }



    @WriteOperation
    public void busNoveRefresh() {
        publish(new NoveRefreshRemoteApplicationEvent(this, getInstanceId(),null));
    }

    //此处原来的destination 参数必须写成arg0 否则就不生效，恶心人，内部机制，这样处理最简单
    @WriteOperation
    public void busRefreshWithDestination(@Selector String arg0) {
        publish(new NoveRefreshRemoteApplicationEvent(this, getInstanceId(), arg0));
    }
}

复制代码

SO 到目前为止一个可用于生产的AB-TEST 机制就实现了，有的同窗就会说了，你这根据数据在进行if else 的逻辑判断调用真是恶心到人了。确实，由于初版目前是最简单的实现，在第二版将进行动态调用的优化。其实动态调用说来也很简单，无非经过定义的接口实现各个不一样的逻辑，而后针对其进行简单的动态代理便可。后期源码会同步上传。

总结： 自定义AB-TEST组件的过程

  • 一、自定义内部端点暴露动态配置刷新，发送刷新事件
  • 二、接收事件，定义刷新策略，只刷新须要的配置便可
  • 三、定义启动初始化方式
  • 四、经过动态代理实现动态逻辑调用（待完成）

qrcode_for_gh_13314ac27929_258.jpg
相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程