Soul网关中利用HTTP长轮询实现数据同步，这也太好用了叭！

Myth丶恋晨 2023-01-23 04:53 83阅读 0赞

**点击关注公众号，利用****碎片时间学习****![b2cc01ee32befe5086096f9cd528cc0b.png][]**

## 前言 ##

Soul是一款异步的、高性能的、跨语言的、响应式的API网关，本文主要来解析Soul中基于Http长轮询同步数据的设计，下图展示了Soul数据同步的流程：

![a4a4a1fa2d88c43ab2824aaf21403208.png][]

Soul主要由soul-admin和soul-web两个核心模块组成，soul-admin是路由和其他信息配置的管理后台；soul-web是网关主体，用于实现路由转发等功能。

soul-web实现了动态配置功能，会同步soul-admin中的路由和其他配置信息，为了提供更高的响应速度，soul-web所有的缓存配置都存在JVM的ConcurrentMap中，每次请求都走的本地缓存，速度非常快

soul-admin在用户发生配置变更之后，会通过EventPublisher发出配置变更通知，由EventDispatcher处理该变更通知，然后根据配置的同步策略（Http、Websocket、Zookeeper），将配置发送给对应的事件处理器

**Http同步策略实现如下：**

soul-web主动发起长轮询请求，默认有90s超时时间，如果soul-admin没有数据变更，则会阻塞Http请求，如果有数据发生变更则响应变更的数据信息，如果超过60s仍然没有数据变更则响应空数据，网关层接到响应后，继续发起Http请求，反复同样的请求

## 启用Http同步策略 ##

soul-admin中开启Http同步策略

soul:
      sync:
        http:
          enabled: true

soul-bootstrap中配置管理后台的请求地址

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

## 源码解析 ##

这块主要是在soul-admin子模块中实现的

## DataChangedListener体系 ##

配置类DataSyncConfiguration中初始化了关于Http长轮询的Bean

@Configuration
    public class DataSyncConfiguration {
    
        @Configuration
        @ConditionalOnProperty(name = "soul.sync.http.enabled", havingValue = "true")
        @EnableConfigurationProperties(HttpSyncProperties.class)
        static class HttpLongPollingListener {
    
            @Bean
            @ConditionalOnMissingBean(HttpLongPollingDataChangedListener.class)
            public HttpLongPollingDataChangedListener httpLongPollingDataChangedListener(final HttpSyncProperties httpSyncProperties) {
                return new HttpLongPollingDataChangedListener(httpSyncProperties);
            }
        }

`HttpLongPollingDataChangedListener`继承自`AbstractDataChangedListener`，和其他同步策略一样都实现自接口`DataChangedListener`

![9aed5c89178cc6ee4f9ea6507d766fde.png][]

`DataChangedListener`这个接口提供了众多不同数据类型变动的方法，供`DataChangedEventDispatcher`调用，`DataChangedEventDispatcher`用于处理数据同步的事件分类及分发

public interface DataChangedListener {
    
        default void onAppAuthChanged(List<AppAuthData> changed, DataEventTypeEnum eventType) {
        }
    
        default void onPluginChanged(List<PluginData> changed, DataEventTypeEnum eventType) {
        }
    
        default void onSelectorChanged(List<SelectorData> changed, DataEventTypeEnum eventType) {
        }
    
        default void onMetaDataChanged(List<MetaData> changed, DataEventTypeEnum eventType) {
    
        }
    
        default void onRuleChanged(List<RuleData> changed, DataEventTypeEnum eventType) {
        }
    
    }

@Component
    public class DataChangedEventDispatcher implements ApplicationListener<DataChangedEvent>, InitializingBean {
    
        private ApplicationContext applicationContext;
    
        //持有DataChangedListener集合
        private List<DataChangedListener> listeners;
    
        public DataChangedEventDispatcher(final ApplicationContext applicationContext) {
            this.applicationContext = applicationContext;
        }
    
        //事件变动时,通知DataChangedListener的不同事件类型的方法
        @Override
        @SuppressWarnings("unchecked")
        public void onApplicationEvent(final DataChangedEvent event) {
            for (DataChangedListener listener : listeners) {
                switch (event.getGroupKey()) {
                    case APP_AUTH:
                        listener.onAppAuthChanged((List<AppAuthData>) event.getSource(), event.getEventType());
                        break;
                    case PLUGIN:
                        listener.onPluginChanged((List<PluginData>) event.getSource(), event.getEventType());
                        break;
                    case RULE:
                        listener.onRuleChanged((List<RuleData>) event.getSource(), event.getEventType());
                        break;
                    case SELECTOR:
                        listener.onSelectorChanged((List<SelectorData>) event.getSource(), event.getEventType());
                        break;
                    case META_DATA:
                        listener.onMetaDataChanged((List<MetaData>) event.getSource(), event.getEventType());
                        break;
                    default:
                        throw new IllegalStateException("Unexpected value: " + event.getGroupKey());
                }
            }
        }
    
        @Override
        public void afterPropertiesSet() {
            Collection<DataChangedListener> listenerBeans = applicationContext.getBeansOfType(DataChangedListener.class).values();
            this.listeners = Collections.unmodifiableList(new ArrayList<>(listenerBeans));
        }
    }

`AbstractDataChangedListener`就是定义了一个模板，让子类可以按照指定步骤进行工作，具体每个步骤的工作细节可以由子类自己实现（模版方法模式），以`onPluginChanged()`插件数据变动方法为例

public abstract class AbstractDataChangedListener implements DataChangedListener, InitializingBean {
      
        @Override
        public void onPluginChanged(final List<PluginData> changed, final DataEventTypeEnum eventType) {
            if (CollectionUtils.isEmpty(changed)) {
                return;
            }
            this.updatePluginCache();
            this.afterPluginChanged(changed, eventType);
        }
      
       //修改缓存(可重写)
        protected void updatePluginCache() {
            this.updateCache(ConfigGroupEnum.PLUGIN, pluginService.listAll());
        }
      
       //自定义结束数据变动后要干什么(可重写)
        protected void afterPluginChanged(final List<PluginData> changed, final DataEventTypeEnum eventType) {
        }

## HttpLongPollingDataChangedListener长轮询实现 ##

soul-admin通过`ConfigController`暴露Http路径供网关调用并监听数据变化

这里调用了`HttpLongPollingDataChangedListener`的`doLongPolling()`方法，`doLongPolling()`是实现Http长轮询的关键：

public class HttpLongPollingDataChangedListener extends AbstractDataChangedListener {
      
     public void doLongPolling(final HttpServletRequest request, final HttpServletResponse response) {
            //因为soul-web可能未收到某个配置变更的通知,因此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;
            }
            //Servlet3.0异步响应http请求
            final AsyncContext asyncContext = request.startAsync();
            asyncContext.setTimeout(0L);
            scheduler.execute(new LongPollingClient(asyncContext, clientIp, HttpConstants.SERVER_MAX_HOLD_TIMEOUT));
        }
      
        class LongPollingClient implements Runnable {
          
            @Override
            public void run() {
                //加入定时任务,如果60s之内没有配置变更,则60s后执行,响应http请求
                this.asyncTimeoutFuture = scheduler.schedule(() -> {
                    clients.remove(LongPollingClient.this);
                    List<ConfigGroupEnum> changedGroups = compareChangedGroup((HttpServletRequest) asyncContext.getRequest());
                    sendResponse(changedGroups);
                }, timeoutTime, TimeUnit.MILLISECONDS);
                clients.add(this);
            }
          
            void sendResponse(final List<ConfigGroupEnum> changedGroups) {
                //取消调度
                if (null != asyncTimeoutFuture) {
                    asyncTimeoutFuture.cancel(false);
                }
                generateResponse((HttpServletResponse) asyncContext.getResponse(), changedGroups);
                //同步完成结束阻塞
                asyncContext.complete();
            }

Http请求到达sou-admin之后，并非立马响应数据，而是利用Servlet3.0的[异步机制][Link 1]，异步响应数据。首先，将长轮询请求任务`LongPollingClient`扔到`BlockingQueue`中，并且开启调度任务，60s后执行，这样做的目的是60s后将该长轮询请求移除队列，即便是这段时间内没有发生配置数据变更。因为即便是没有配置变更，也得让网关知道，总不能让其干等吧，而且网关请求配置服务时，也有90s的超时时间

## 长轮询时配置数据变更立即响应数据 ##

讲完了Http请求到达sou-admin这段时间没有变更配置数据60s后返回响应，如果这段时间内，存在配置数据变更呢？

首先Http请求到达sou-admin时最终拿到的配置信息数据都是从`AbstractDataChangedListener`的`Cache`中获取的

public abstract class AbstractDataChangedListener implements DataChangedListener, InitializingBean {
    
        protected static final ConcurrentMap<String, ConfigDataCache> CACHE = new ConcurrentHashMap<>();

那`AbstractDataChangedListener`中本地缓存数据是何时更新的，只有本地缓存更新了才会立即响应数据

**AbstractDataChangedListener中本地缓存数据更新时机：**

*  触发DataChangedEvent事件，DataChangedEventDispatcher监听到来处理数据同步的事件分类及分发
 *  定时任务同步

**1）触发Spring事件，DataChangedEventDispatcher监听到来处理数据同步的事件分类及分发**

就是`DataChangedListener`体系中讲解的内容，下面以创建和修改Plugin为例：

@Service("pluginService")
    public class PluginServiceImpl implements PluginService {    
    
      @Override
        @Transactional(rollbackFor = Exception.class)
        public String createOrUpdate(final PluginDTO pluginDTO) {
            final String msg = checkData(pluginDTO);
            if (StringUtils.isNoneBlank(msg)) {
                return msg;
            }
            PluginDO pluginDO = PluginDO.buildPluginDO(pluginDTO);
            DataEventTypeEnum eventType = DataEventTypeEnum.CREATE;
            if (StringUtils.isBlank(pluginDTO.getId())) {
                insertPluginDataToResource(pluginDTO);
                pluginMapper.insertSelective(pluginDO);
            } else {
                eventType = DataEventTypeEnum.UPDATE;
                pluginMapper.updateSelective(pluginDO);
            }
    
            //触发事件
            eventPublisher.publishEvent(new DataChangedEvent(ConfigGroupEnum.PLUGIN, eventType,
                    Collections.singletonList(PluginTransfer.INSTANCE.mapToData(pluginDO))));
            return StringUtils.EMPTY;
        }

创建和修改Plugin之后，触发了`DataChangedEvent`，`DataChangedEventDispatcher`根据groupKey进行分发，调用了`AbstractDataChangedListener中`实现的`onXxxChanged()`方法：

@Component
    public class DataChangedEventDispatcher implements ApplicationListener<DataChangedEvent>, InitializingBean {    
    
      //事件变动时,通知DataChangedListener的不同事件类型的方法
        @Override
        @SuppressWarnings("unchecked")
        public void onApplicationEvent(final DataChangedEvent event) {
            for (DataChangedListener listener : listeners) {
                switch (event.getGroupKey()) {
                    case APP_AUTH:
                        listener.onAppAuthChanged((List<AppAuthData>) event.getSource(), event.getEventType());
                        break;
                    case PLUGIN:
                        listener.onPluginChanged((List<PluginData>) event.getSource(), event.getEventType());
                        break;
                    case RULE:
                        listener.onRuleChanged((List<RuleData>) event.getSource(), event.getEventType());
                        break;
                    case SELECTOR:
                        listener.onSelectorChanged((List<SelectorData>) event.getSource(), event.getEventType());
                        break;
                    case META_DATA:
                        listener.onMetaDataChanged((List<MetaData>) event.getSource(), event.getEventType());
                        break;
                    default:
                        throw new IllegalStateException("Unexpected value: " + event.getGroupKey());
                }
            }
        }

`AbstractDataChangedListener`中采用了模版方法模式，`onXxxChanged()`方法中修改了缓存，提供了`afterXxxChanged()`方法让子类自定义结束数据变动后要干什么

`HttpLongPollingDataChangedListener`中重写了`AbstractDataChangedListener`中的`afterXxxChanged()`方法，`afterXxxChanged()`方法是通过事件变更触发长轮询立即响应的关键

public class HttpLongPollingDataChangedListener extends AbstractDataChangedListener {
      
      @Override
        protected void afterPluginChanged(final List<PluginData> changed, final DataEventTypeEnum eventType) {
            scheduler.execute(new DataChangeTask(ConfigGroupEnum.PLUGIN));
        }
      
        class DataChangeTask implements Runnable {
    
            private final ConfigGroupEnum groupKey;
    
            private final long changeTime = System.currentTimeMillis();
    
            DataChangeTask(final ConfigGroupEnum groupKey) {
                this.groupKey = groupKey;
            }
    
            @Override
            public void run() {
               //soul-admin发生了配置变更,挨个将队列中的请求移除,并予以响应
                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);
                }
            }
        }

如果这段时间内，管理员变更了配置数据，此时，会挨个移除队列中的长轮询请求，并响应数据，告知是哪个Group的数据发生了变更（将插件、规则、流量配置、用户配置数据分成不同的组）。

[网关][Link 2]收到响应信息之后，只知道是哪个Group发生了配置变更，还需要再次请求该Group的配置数据，这时候会调用ConfigController中暴露的fetch接口拿到最新的配置数据

@ConditionalOnBean(HttpLongPollingDataChangedListener.class)
    @RestController
    @RequestMapping("/configs")
    @Slf4j
    public class ConfigController {
    
        @Resource
        private HttpLongPollingDataChangedListener longPollingListener;
    
        @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);
        }

关于为什么不是直接将变更的数据写出，作者给出的回答是：因为Http长轮询机制只能保证准实时，如果在网关层处理不及时，或者管理员频繁更新配置，很有可能便错过了某个配置变更的推送，安全起见，我们只告知某个Group信息发生了变更。

**2）定时任务同步**

定时任务同步是因为触发事件只是单个节点完成缓存更新，假设soul-admin存在A、B两个节点，更新配置数据的请求到达A节点完成了A节点内本地缓存数据的更新，而B节点并未更新本地缓存，所以B节点的数据是通过定时任务来同步的

定时任务同步也是在`HttpLongPollingDataChangedListener`中实现的

public class HttpLongPollingDataChangedListener extends AbstractDataChangedListener {
       
      @Override
        protected void afterInitialize() {
            long syncInterval = httpSyncProperties.getRefreshInterval().toMillis();
            //定期检查数据的更改并更新缓存
            scheduler.scheduleWithFixedDelay(() -> {
                log.info("http sync strategy refresh config start.");
                try {
                    this.refreshLocalCache();
                    log.info("http sync strategy refresh config success.");
                } catch (Exception e) {
                    log.error("http sync strategy refresh config error!", e);
                }
            }, syncInterval, syncInterval, TimeUnit.MILLISECONDS);
            log.info("http sync strategy refresh interval: {}ms", syncInterval);
        }
      
        private void refreshLocalCache() {
            this.updateAppAuthCache();
            this.updatePluginCache();
            this.updateRuleCache();
            this.updateSelectorCache();
            this.updateMetaDataCache();
        }
      
       //最终调用的更新cache的方法
        protected <T> void updateCache(final ConfigGroupEnum group, final List<T> data) {
            String json = GsonUtils.getInstance().toJson(data);
            ConfigDataCache newVal = new ConfigDataCache(group.name(), json, Md5Utils.md5(json), System.currentTimeMillis());
            ConfigDataCache oldVal = CACHE.put(newVal.getGroup(), newVal);
            log.info("update config cache[{}], old: {}, updated: {}", group, oldVal, newVal);
        }

当定时任务更新配置数据之后，md5的值会改变，无论`doLongPolling()`方法中开启异步响应http请求之前或是60s内没有事件触发配置变更LongPollingClient返回响应其中都会调用`compareChangedGroup()`方法，判断一下缓存中md5的值是否被改变，其实就是针对定时任务更新配置数据时，也能返回哪个Group的数据发生了变更

private List<ConfigGroupEnum> compareChangedGroup(final HttpServletRequest request) {
        List<ConfigGroupEnum> changedGroup = new ArrayList<>(ConfigGroupEnum.values().length);
        for (ConfigGroupEnum group : ConfigGroupEnum.values()) {
            // md5,lastModifyTime
            String[] params = StringUtils.split(request.getParameter(group.name()), ',');
            if (params == null || params.length != 2) {
                throw new SoulException("group param invalid:" + request.getParameter(group.name()));
            }
            String clientMd5 = params[0];
            long clientModifyTime = NumberUtils.toLong(params[1]);
            ConfigDataCache serverCache = CACHE.get(group.name());
            // do check.
            if (this.checkCacheDelayAndUpdate(serverCache, clientMd5, clientModifyTime)) {
                changedGroup.add(group);
            }
        }
        return changedGroup;
    }

## 总结 ##

![43c133132f417f374b4d1cd0e825107e.png][]

*  后台通过Controller层暴露API给网关，网关请求后台时后台并不是立即返回响应（数据有无变化），而是hold住请求最大60秒的时间。这些被hold住的请求会加入到阻塞队列中作为内存缓存
 *  这60秒钟如果有事件直接触发的数据变化，通过DataChangedEventDispatcher分发到的HttpLongPollingDataChangedListener，则立即调用线程池在阻塞队列中遍历所有被hold住的请求，塞入响应信息并释放掉
 *  如果60秒过后依然没有数据变化，hold住的请求会被释放，且阻塞队列的对应请求对象被剔除
 *  在开启异步响应Http请求之前或是60s内没有事件触发配置变更返回响应对比缓存中md5的值和网关请求中携带的md5的值看是否是定时任务触发了缓存的变更，从而返回哪个Group的数据发生了变更

Soul网关中的Http长轮询只是Http长轮询中的一种实现，很大程度上参考了Nacos配置中心的设计，在Apollo中用另一种API同样实现长轮询的机制。另外，欢迎关注公众号Java笔记虾，后台回复“后端面试”，送你一份面试题宝典！

## 参考： ##

*  https://dromara.org/zh/projects/soul/data-sync/
 *  https://dromara.org/zh/blog/soul\_source\_learning\_08\_httplongpolling\_01/
 *  https://dromara.org/zh/blog/soul\_source\_learning\_09\_httplongpolling\_02/

*（感谢阅读，希望对你所有帮助)*

*来源：blog.csdn.net/qq\_40378034/article/details/113927834*

推荐好文

[强大，10k+点赞的 SpringBoot 后台管理系统竟然出了详细教程！][10k_ SpringBoot]

[分享一套基于SpringBoot和Vue的企业级中后台开源项目，代码很规范！][SpringBoot_Vue]

[能挣钱的，开源 SpringBoot 商城系统，功能超全，超漂亮][SpringBoot]![b0c7bdcefe1ca73a47fd0694abb9ed3d.png][]

[b2cc01ee32befe5086096f9cd528cc0b.png]: /images/20221004/5d3c034f5dbd4a0eb30788bd6675a234.png
[a4a4a1fa2d88c43ab2824aaf21403208.png]: /images/20221004/f7281663af98498bb82cf25d49649947.png
[9aed5c89178cc6ee4f9ea6507d766fde.png]: /images/20221004/35bedb47ede240fb9968b11ddfa4bbff.png
[Link 1]: http://mp.weixin.qq.com/s?__biz=MzU2MTI4MjI0MQ%3D%3D&chksm=fc79a221cb0e2b37fe94a30ed0f5085ba386400e2064b3e6973afde6d4b017828ff4c643f84f&idx=2&mid=2247503759&scene=21&sn=5c4096f05da4c1b18a06badf17bb07f3#wechat_redirect
[Link 2]: http://mp.weixin.qq.com/s?__biz=MzU2MTI4MjI0MQ%3D%3D&chksm=fc79af45cb0e2653707794498e511335540103a297079ca088f982e4d1f7b86b687d2ebe8fb2&idx=3&mid=2247500523&scene=21&sn=cee5cb94811d32db09267bfab004abba#wechat_redirect
[43c133132f417f374b4d1cd0e825107e.png]: /images/20221004/df5d1adfe01543508db3c407ff34dbe0.png
[10k_ SpringBoot]: http://mp.weixin.qq.com/s?__biz=MzI4Njc5NjM1NQ%3D%3D&chksm=ebd5c411dca24d07446a4f22772cd12d2cc1143b62eb922a14500ae981cd57fb850ef74ca257&idx=2&mid=2247498557&scene=21&sn=84902d7030e174e152615d8fae3755cc#wechat_redirect
[SpringBoot_Vue]: http://mp.weixin.qq.com/s?__biz=MzU2MTI4MjI0MQ%3D%3D&chksm=fc799be6cb0e12f0b58479f8b031bc5ae79efdca1880ff9b19f0a107c3a88c37b9596dc73203&idx=2&mid=2247497288&scene=21&sn=65d7111a19e920df1e9416959594b2f4#wechat_redirect
[SpringBoot]: http://mp.weixin.qq.com/s?__biz=MzU2MTI4MjI0MQ%3D%3D&chksm=fc799a9ccb0e138a64a2671f31b8d4be1b3c061a288f7b0d1d4af02a60c45a72231446b91e00&idx=2&mid=2247497522&scene=21&sn=c5b4e596be4ac7d7f293e7e2f0ea0769#wechat_redirect
[b0c7bdcefe1ca73a47fd0694abb9ed3d.png]: /images/20221004/7a84c84451004a27a07affbc600ab019.png