目标

研究Soul网关在启动的过程都做了些什么：

• SoulConfiguration被spring容器加载，主要是初始化以下一些Bean
  1. 初始化SoulWebHandler：负责处理http请求
  2. 初始化DispatcherHandler
  3. 初始化PluginDataSubscriber：负责将插件、选择器、规则等配置数据同步到内存中
  4. 初始化DubboParamResolveService：如果pom.xml配置dubbo plugin则进行DubboParamResolveService的初始化
  5. 初始化CrossFilter：跨域请求
• 同步数据，这次是基于http长轮询技术实现同步数据

系统配置

首先，在pom.xml里要加上如下配置：

<dependency>
    <groupId>org.dromara</groupId>
    <artifactId>soul-spring-boot-starter-gateway</artifactId>
    <version>${project.version}</version>
</dependency>

<dependency>
    <groupId>org.dromara</groupId>
    <artifactId>soul-spring-boot-starter-sync-data-http</artifactId>
    <version>${project.version}</version>
</dependency>

在application.yml配置：

soul :
  sync:
    http:
      url : http://localhost:9095

注意，soul Admin也要配置

soul :
  sync:
    http:
      enabled: true

源码分析

soul-spring-boot-starter-gateway是一个spring-boot-starter，它负责在spring容器启动时加载一些配置信息，如@Configuration或@Bean注解相关的Bean。我们可以看下spring.factories文件（注：在soul-web模块的resources里）

# Auto Configure
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
org.dromara.soul.web.configuration.SoulConfiguration
org.springframework.context.ApplicationListener=\
org.dromara.soul.web.logo.SoulLogo

spring容器启动时，会初始化SoulConfiguration和HttpSyncDataConfiguration。主要功能创建一些Spring Bean。

SoulConfiguration.java

/**
 * Init SoulWebHandler.
 * 创建SoulWebHandler类，并放入spring bean容器中
 * SoulWebHandler 这个类是用来处理web请求的。
 *
 * @param plugins this plugins is All impl SoulPlugin.
 * @return {@linkplain SoulWebHandler}
 */
@Bean("webHandler")
public SoulWebHandler soulWebHandler(final ObjectProvider<List<SoulPlugin>> plugins) {
    List<SoulPlugin> pluginList = plugins.getIfAvailable(Collections::emptyList);
    final List<SoulPlugin> soulPlugins = pluginList.stream()
            .sorted(Comparator.comparingInt(SoulPlugin::getOrder)).collect(Collectors.toList());
    soulPlugins.forEach(soulPlugin -> log.info("load plugin:[{}] [{}]", soulPlugin.named(), soulPlugin.getClass().getName()));
    return new SoulWebHandler(soulPlugins);
}

/**
 * Plugin data subscriber plugin data subscriber.
 * 创建CommonPluginDataSubscriber类，
 * 主要功能是将Plugin、Selector、Rule信息同步网关内存中
 *
 * @param pluginDataHandlerList the plugin data handler list
 * @return the plugin data subscriber
 */
@Bean
public PluginDataSubscriber pluginDataSubscriber(final ObjectProvider<List<PluginDataHandler>> pluginDataHandlerList) {
    return new CommonPluginDataSubscriber(pluginDataHandlerList.getIfAvailable(Collections::emptyList));
}

/**
 * Generic param resolve service dubbo param resolve service.
 * 负责将http请求的参数转化成dubbo服务的参数
 *
 * @return the dubbo param resolve service
 */
@Bean
@ConditionalOnMissingBean(value = DubboParamResolveService.class, search = SearchStrategy.ALL)
public DubboParamResolveService defaultDubboParamResolveService() {
    return new DefaultDubboParamResolveServiceImpl();
}

// 如果在application.yml里配置soul.cross.enabled=true，则创建CrossFilter
@ConditionalOnProperty(name = "soul.cross.enabled", havingValue = "true")
public WebFilter crossFilter() {
    return new CrossFilter();
}

// 加载application.yml中soul的配置信息
@ConfigurationProperties(prefix = "soul")
public SoulConfig soulConfig() {
    return new SoulConfig();
}

HttpSyncDataConfiguration.java

Spring容器加裁HttpSyncDataConfiguration配置的前置条件是当前classpath下存在HttpSyncDataService类和在配置文件有soul.sync.http.ul的信息

// HttpSyncData相关配置:
// Soul Admin URL
// connection timeount
// delay time 
@Bean
@ConfigurationProperties(prefix = "soul.sync.http")
public HttpConfig httpConfig() {
    return new HttpConfig();
}

// 初始化HttpSyncDataService
@Bean
public SyncDataService httpSyncDataService(final ObjectProvider<HttpConfig> httpConfig,
                                           final ObjectProvider<PluginDataSubscriber> pluginSubscriber,
                                       final ObjectProvider<List<MetaDataSubscriber>> metaSubscribers,
                                           final ObjectProvider<List<AuthDataSubscriber>> authSubscribers) {
    log.info("you use http long pull sync soul data");
    return new HttpSyncDataService(Objects.requireNonNull(httpConfig.getIfAvailable()), Objects.requireNonNull(pluginSubscriber.getIfAvailable()),
            metaSubscribers.getIfAvailable(Collections::emptyList), authSubscribers.getIfAvailable(Collections::emptyList));
}

HttpSyncDataService.java

HttpSyncDataService是http长轮询同步数据的核心类

// 在创建对象时，调用该方法
private void start() {
    // It could be initialized multiple times, so you need to control that.
    // 将RUNNING的状态设置为ture。基于CAS原理，防止并发执行线程池的创建
    if (RUNNING.compareAndSet(false, true)) {
        // fetch all group configs.
        // 调用Soul Admin http api 获取配置数据
        this.fetchGroupConfig(ConfigGroupEnum.values());
        // 设置线程池的线程数量，serverList是配置了多个soul Admin的地址。
        int threadSize = serverList.size();
        // 创建线程池，无限大小的任务队列。
        this.executor = new ThreadPoolExecutor(threadSize, threadSize, 60L, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(),
                // 设置线程池name，并且线程是后台线程
                SoulThreadFactory.create("http-long-polling", true));
        // start long polling, each server creates a thread to listen for changes.
        // 为每个Soul Admion的地址，创建一个工作任务。
        this.serverList.forEach(server -> this.executor.execute(new HttpLongPollingTask(server)));
    } else {
        log.info("soul http long polling was started, executor=[{}]", executor);
    }
}

// 创建任务，负责从Soul Admin拉取配置数据
class HttpLongPollingTask implements Runnable {
    // Soul Admin 地址
    private String server;
    // 重试次数
    private final int retryTimes = 3;

    HttpLongPollingTask(final String server) {
        this.server = server;
    }

    @Override
    public void run() {
        // 如果RUNNING 为true，就一直循环
        while (RUNNING.get()) {
            // 重试
            for (int time = 1; time <= retryTimes; time++) {
                try {
                    // 获取是否有变更的配置数，如果没有则不处理
                    doLongPolling(server);
                } catch (Exception e) {
                    // print warnning log.
                    // 如果获取配置数据出现异常，当前次数小于重试次数，则sleep 5秒
                    if (time < retryTimes) {
                        log.warn("Long polling failed, tried {} times, {} times left, will be suspended for a while! {}",
                                time, retryTimes - time, e.getMessage());
                        ThreadUtils.sleep(TimeUnit.SECONDS, 5);
                        continue;
                    }
                    // print error, then suspended for a while.
                    log.error("Long polling failed, try again after 5 minutes!", e);
                    // 如果当前次数大于等于重试次数，则sleep 5分钟。可能Soul Admin服务不可用
                    ThreadUtils.sleep(TimeUnit.MINUTES, 5);
                }
            }
        }
        log.warn("Stop http long polling.");
    }
}

private void doLongPolling(final String server) {
    // ...
    try {
        // 调用Soul Admin请求，获取配置数据。注意如果没有变更的数据，则阻塞60s（是Soul Admin控制时长）。
        // 如有变更数据立即返回
        String json = this.httpClient.postForEntity(listenerUrl, httpEntity, String.class).getBody();
        // ...
    } catch (RestClientException e) {
        String message = String.format("listener configs fail, server:[%s], %s", server, e.getMessage());
        throw new SoulException(message, 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);
        }
    }
}

private void doFetchGroupConfig(final String server, final ConfigGroupEnum... groups) {
   // ...
    try {
        // 从Soul Admin获取配置数据
        json = this.httpClient.getForObject(url, String.class);
    } catch (RestClientException e) {
        String message = String.format("fetch config fail from server[%s], %s", url, e.getMessage());
        log.warn(message);
        throw new SoulException(message, e);
    }
    // update local cache 更新到内存中，
    boolean updated = this.updateCacheWithJson(json);
    if (updated) {
        log.info("get latest configs: [{}]", json);
        return;
    }
    // 如果没有更新，则sleep 30s ...
    ThreadUtils.sleep(TimeUnit.SECONDS, 30);
}

Soul网关同步数据的时候，也就是调用 http://localhost:9095/configs/listener 时，如果没有变更的数据会阻塞60s。这个逻辑在Soul Admin端是怎么实现的呢？我们继续分析。我们看Soul Admin的源码。

HttpLongPollingDataChangedListener

Soul网关调用 http://localhost:9095/configs/listener 时，会调用HttpLongPollingDataChangedListener类的doLongPolling方法

public void doLongPolling(final HttpServletRequest request, final HttpServletResponse response) {
    // compare group md5
    // 当有配置数据有变更时，会写入到内存中并且生成新的md5值。
    // 如果客户端传过来的md5值和新的md5值不匹配则返回返回需要更新的配置项。
    List<ConfigGroupEnum> changedGroup = compareChangedGroup(request);
    // 获取客户端的IP，区分不同的Soul网关
    String clientIp = getRemoteIp(request);
    // response immediately.
    // 如果要变更的配置不为空，则立即返回配置数据。
    if (CollectionUtils.isNotEmpty(changedGroup)) {
        this.generateResponse(response, changedGroup);
        log.info("send response with the changed group, ip={}, group={}", clientIp, changedGroup);
        return;
    }
    // listen for configuration changed.
    final AsyncContext asyncContext = request.startAsync();
    // AsyncContext.settimeout() does not timeout properly, so you have to control it yourself
    asyncContext.setTimeout(0L);
    // block client's thread.
    // 会阻塞客户端的线程
    scheduler.execute(new LongPollingClient(asyncContext, clientIp, HttpConstants.SERVER_MAX_HOLD_TIMEOUT));
}

class LongPollingClient implements Runnable {
    @Override
    public void run() {
        // delay 设置 为 60s HttpConstants.SERVER_MAX_HOLD_TIMEOUT
        // 延迟 60s 后执行
        this.asyncTimeoutFuture = scheduler.schedule(() -> {
            clients.remove(LongPollingClient.this);
            List<ConfigGroupEnum> changedGroups = compareChangedGroup((HttpServletRequest) asyncContext.getRequest());
            // 如果还没有变更的配置数据，则返回空
            sendResponse(changedGroups);
        }, timeoutTime, TimeUnit.MILLISECONDS);
        clients.add(this);
    }    
}

// 当有配置数据发生变更时，会创建该任务
class DataChangeTask implements Runnable {
    @Override
    public void run() {
        // 遍历所有客户端
        for (Iterator<LongPollingClient> iter = clients.iterator(); iter.hasNext();) {
            LongPollingClient client = iter.next();
            iter.remove();
            // 向客户端返回Response（变更后的配置数据）
            client.sendResponse(Collections.singletonList(groupKey));
            log.info("send response with the changed group,ip={}, group={}, changeTime={}", client.ip, groupKey, changeTime);
        }
    }
}

AbstractDataChangedListener

该类采用监听器模式，当有配置数据发变更时，会触发相应方法。
如MetaData数据发生变更时

// 当有MetaData数据发生变更时，会触发该方法
@Override
public void onMetaDataChanged(final List<MetaData> changed, final DataEventTypeEnum eventType) {
    if (CollectionUtils.isEmpty(changed)) {
        return;
    }
    // 更新缓存（内存中）
    this.updateMetaDataCache();
    // 创建DataChangeTask，向client发送变更的数据
    this.afterMetaDataChanged(changed, eventType);
}

总结

• 了解Soul网关在启动时会初始化哪些Spring Bean
• 了解Soul网关基于http长轮徇进行数据同步的实现。采取pull的模式
• 了解监听器设计模式的使用