【Soul网关探秘】http数据同步-变更通知机制

引言

上一篇，梳理除了

soul-admin

在发出数据变更通知前的处理脉络，本篇开始探究 http 同步策略的变更通知机制，

不同数据变更的通知机制应当是一致的，故本篇以 selector 配置变更通知为切入点进行深入。

配置操作入口

找到 ConfigController，这是配置操作的入口

【Soul网关探秘】http数据同步-变更通知机制

其持有一个 HttpLongPollingDataChangedListener 引用，通过 HttpLongPollingDataChangedListener 实现配置变更通知订阅和配置获取。

通知订阅：

@PostMapping(value = "/listener")
public void listener(final HttpServletRequest request, final HttpServletResponse response) {
    longPollingListener.doLongPolling(request, response);
}

配置获取：

@GetMapping("/fetch")
public SoulAdminResult fetchConfigs(@NotNull final String[] groupKeys) {
    Map<String, ConfigData<?>> result = Maps.newHashMap();
    for (String groupKey : groupKeys) {
        ConfigData<?> data = longPollingListener.fetchConfig(ConfigGroupEnum.valueOf(groupKey));
        result.put(groupKey, data);
    }
    return SoulAdminResult.success(SoulResultMessage.SUCCESS, result);
}

通知订阅实现

使用 HttpLongPollingDataChangedListener#doLongPolling 实现通知订阅

public void doLongPolling(final HttpServletRequest request, final HttpServletResponse response) {
    // 比较配置组md5
    List<ConfigGroupEnum> changedGroup = compareChangedGroup(request);
    String clientIp = getRemoteIp(request);
    // 发现配置组变化则立即响应
    if (CollectionUtils.isNotEmpty(changedGroup)) {
        this.generateResponse(response, changedGroup);
        log.info("send response with the changed group, ip={}, group={}", clientIp, changedGroup);
        return;
    }
    // 监听配置变化
    final AsyncContext asyncContext = request.startAsync();
    asyncContext.setTimeout(0L);
    // 阻塞客户端线程
    scheduler.execute(new LongPollingClient(asyncContext, clientIp, HttpConstants.SERVER_MAX_HOLD_TIMEOUT));
}

通过比较 MD5 检查配置组是否发生变更，若配置组发生变更则立即响应，否则阻塞客户端线程。

此处 compareChangedGroup 实现不做深究，继续看LongPollingClient 具体处理：

@Override
public void run() {
    this.asyncTimeoutFuture = scheduler.schedule(() -> {
        clients.remove(LongPollingClient.this);
        List<ConfigGroupEnum> changedGroups = compareChangedGroup((HttpServletRequest) asyncContext.getRequest());
        sendResponse(changedGroups);
    }, timeoutTime, TimeUnit.MILLISECONDS);
    clients.add(this);
}

这里将 client 加入 clients 的同时，开启了一个定时任务，负责超时移除 client 并返回发生变化的配置组信息。

超时时间为构造时传入的 HttpConstants.SERVER_MAX_HOLD_TIMEOUT = 60s

配置获取实现

使用 AbstractDataChangedListener#fetchConfig 实现配置获取

public ConfigData<?> fetchConfig(final ConfigGroupEnum groupKey) {
    ConfigDataCache config = CACHE.get(groupKey.name());
    switch (groupKey) {
        case APP_AUTH:
            ...
        case PLUGIN:
            ...
        case RULE:
            ...
        case SELECTOR:
            List<SelectorData> selectorList = GsonUtils.getGson().fromJson(config.getJson(), new TypeToken<List<SelectorData>>() {
            }.getType());
            return new ConfigData<>(config.getMd5(), config.getLastModifyTime(), selectorList);
        case META_DATA:
            ...
        default:
            throw new IllegalStateException("Unexpected groupKey: " + groupKey);
    }
}

这里从 CACHE 缓存获取对应配置组信息，包装成 ConfigData 并返回。

建立订阅关系

soul-web

端通过 HttpSyncDataConfiguration 初始化 HttpSyncDataService 并注入 spring容器。

HttpSyncDataService#start 方法在初始化时完成配置获取和订阅：

private void start() {
    // It could be initialized multiple times, so you need to control that.
    if (RUNNING.compareAndSet(false, true)) {
        // fetch all group configs.
        this.fetchGroupConfig(ConfigGroupEnum.values());
        int threadSize = serverList.size();
        this.executor = new ThreadPoolExecutor(threadSize, threadSize, 60L, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(),
                SoulThreadFactory.create("http-long-polling", true));
        // start long polling, each server creates a thread to listen for changes.
        this.serverList.forEach(server -> this.executor.execute(new HttpLongPollingTask(server)));
    } else {
        log.info("soul http long polling was started, executor=[{}]", executor);
    }
}

1）配置获取

private void fetchGroupConfig(final ConfigGroupEnum... groups) throws SoulException {
    for (int index = 0; index < this.serverList.size(); index++) {
        String server = serverList.get(index);
        try {
            this.doFetchGroupConfig(server, groups);
            break;
        } catch (SoulException e) {
            // no available server, throw exception.
            if (index >= serverList.size() - 1) {
                throw e;
            }
            log.warn("fetch config fail, try another one: {}", serverList.get(index + 1));
        }
    }
}

doFetchGroupConfig 内部发起配置获取请求并更新本地缓存

private void doFetchGroupConfig(final String server, final ConfigGroupEnum... groups) {
    ...
    String url = server + "/configs/fetch?" + StringUtils.removeEnd(params.toString(), "&");
    ...
    try {
        json = this.httpClient.getForObject(url, String.class);
    } catch (RestClientException e) {
        ...
    }
    // update local cache
    boolean updated = this.updateCacheWithJson(json);
    ...
}

2）配置订阅

借助 HttpLongPollingTask 完成

@Override
public void run() {
    while (RUNNING.get()) {
        for (int time = 1; time <= retryTimes; time++) {
            try {
                doLongPolling(server);
            } catch (Exception e) {
                ...
            }
        }
    }
    log.warn("Stop http long polling.");
}

HttpLongPollingTask 不断循环 doLongPolling，此处有 retry 操作

private void doLongPolling(final String server) {
    ...
    String listenerUrl = server + "/configs/listener";
    ...
    try {
        String json = this.httpClient.postForEntity(listenerUrl, httpEntity, String.class).getBody();
        log.debug("listener result: [{}]", json);
        groupJson = GSON.fromJson(json, JsonObject.class).getAsJsonArray("data");
    } catch (RestClientException e) {
        ...
    }
    if (groupJson != null) {
        // fetch group configuration async.
        ConfigGroupEnum[] changedGroups = GSON.fromJson(groupJson, ConfigGroupEnum[].class);
        if (ArrayUtils.isNotEmpty(changedGroups)) {
            log.info("Group config changed: {}", Arrays.toString(changedGroups));
            this.doFetchGroupConfig(server, changedGroups);
        }
    }
}

doLongPolling 内部发起 post 请求订阅配置变更，若发生变更则重新获取配置。

至此，通知订阅处理脉络已清晰：

soul-web 端通过 http 发起订阅请求

soul-admin 端收到请求，通过比较 MD5 检查配置组是否存在变更

若存在变更，则立即响应变更组信息

若无变更，则阻塞客户端线程，并开启定时任务 60s 后重新比较配置组变更并返回响应

soul-web 端收到响应，判断配置组是否存在变更

若存在变更，则发起获取配置请求获取最新配置信息

soul-web 重新发起订阅请求

配置变更

上回我们说到AbstractDataChangedListener 的 onSelectorChanged 实现：

public void onSelectorChanged(final List<SelectorData> changed, final DataEventTypeEnum eventType) {
    if (CollectionUtils.isEmpty(changed)) {
        return;
    }
    // 更新 selector 缓存
    this.updateSelectorCache();
    // selector 变更后处理，实现具体的变更通知
    this.afterSelectorChanged(changed, eventType);
}

这里 selector 变更处理是先更缓存后发通知，继续看 afterSelectorChanged 实现。

HttpLongPollingDataChangedListener 真正实现了 AbstractDataChangedListener 的 afterSelectorChanged：

@Override
protected void afterSelectorChanged(final List<SelectorData> changed, final DataEventTypeEnum eventType) {
    scheduler.execute(new DataChangeTask(ConfigGroupEnum.SELECTOR));
}

由定时任务重复执行 DataChangeTask，DataChangeTask 具体处理如下：

@Override
public void run() {
    for (Iterator<LongPollingClient> iter = clients.iterator(); iter.hasNext();) {
        LongPollingClient client = iter.next();
        iter.remove();
        client.sendResponse(Collections.singletonList(groupKey));
        log.info("send response with the changed group,ip={}, group={}, changeTime={}", client.ip, groupKey, changeTime);
    }
}

DataChangeTask 负责从 clients 依次移除 LongPollingClient 并将 groupKey 作为响应返回，sendResponse 内部处理如下：

void sendResponse(final List<ConfigGroupEnum> changedGroups) {
    // cancel scheduler
    if (null != asyncTimeoutFuture) {
        asyncTimeoutFuture.cancel(false);
    }
    generateResponse((HttpServletResponse) asyncContext.getResponse(), changedGroups);
    asyncContext.complete();
}

负责生成响应报文并异步响应客户端，注意有个 asyncTimeoutFuture.cancel 操作，取消之前的 60s 超时响应。

总结

本篇梳理和分析了

soul-web

端到

soul-admin

端的配置变更通知订阅关系建立过程，配合上配置获取接口，完成了整个 http 数据同步策略的变更通知机制。

下篇，将探究 http 同步策略的web端处理变更通知，期待惊喜。

个人知识库

高性能微服务API网关-Soul

【Soul网关探秘】http数据同步-变更通知机制

引言

配置操作入口

通知订阅实现

配置获取实现

建立订阅关系

配置变更

总结

个人知识库

继续阅读

关于Gradle配置的小结

Java小案例——随机数猜测随机数猜测

nginx location中斜线的位置的重要性

27 Best Free Eclipse Plug-ins for Java Developer to be ProductiveCode Quality PluginsText Editor PluginsDependency ManagementVersion Control Integration PluginsFramework Development Continuous Integration Related PluginsOther Utility Plugins

Java String.format方法的简单使用

neo4j之cypher使用文档

GitHub连夜封杀！这份阿里 10W 字内部 Java 字面试手册到底有多强？

spark/scala关于【资源文件】加载方法概述外部文件加载方案测试资源文件打包入jar包中小结

mybatis_入门程序Mybatis入门

AOP编程_Android优雅权限框架(1)概念基础，2021金三银四前言正文大纲正文

Effective Java 8:通用程序设计

OOM三种类型

工厂模式-三种类型

【递归】高效率求2的n次幂

win10本地scala和spark安装安装scala安装spark

scala (3) Function 和 Method