任务

  1. 启动admin，与网关。 admin操作，使用websocket同步数据到网关

websocket 同步

打开 websocket 同步策略，启动 soul-admin。

1
2
3
4
soul:
  sync:
    websocket:
      enabled: true

配置 websocket 的地址，启动网关。可以看到 websocket 同步策略配置成功：you use websocket sync soul data.......。

1
2
3
4
soul:
  sync:
    websocket :
      urls: ws://localhost:9095/websocket

首先，soul使用 java-websocket 这个第三方库来进行websocket连接。重点可以看一下 soul-sync-data-websocket 工程，其中引入了 java-websocket 的库。主要是通过 WebsocketCollector ，对远程连接进行监控。

1
2
3
4
5
<dependency>
    <groupId>org.java-websocket</groupId>
    <artifactId>Java-WebSocket</artifactId>
    <version>${java-websocket.version}</version>
</dependency>
1
2
3
4
5
@OnOpen
public void onOpen(final Session session) {
    log.info("websocket on open successful....");
    SESSION_SET.add(session);
}

此时，如果 admin 服务停了，网关服务将会输出 websocket reconnection is error 异常日志，每隔 30 秒检查一次；此时重启 admin 服务，网关服务将会输出websocket reconnect is successful，重连成功。其中，主要逻辑集中在 WebsocketSyncDataService 。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
public WebsocketSyncDataService(final WebsocketConfig websocketConfig,
                                    final PluginDataSubscriber pluginDataSubscriber,
                                    final List<MetaDataSubscriber> metaDataSubscribers,
                                    final List<AuthDataSubscriber> authDataSubscribers) {
    ...
    try {
        for (WebSocketClient client : clients) {
            boolean success = client.connectBlocking(3000, TimeUnit.MILLISECONDS);
            if (success) {
                log.info("websocket connection is successful.....");
            } else {
                log.error("websocket connection is error.....");
            }
            executor.scheduleAtFixedRate(() -> {
                try {
                    if (client.isClosed()) {
                        boolean reconnectSuccess = client.reconnectBlocking();
                        if (reconnectSuccess) {
                            log.info("websocket reconnect is successful.....");
                        } else {
                            log.error("websocket reconnection is error.....");
                        }
                    }
                } catch (InterruptedException e) {
                    log.error("websocket connect is error :{}", e.getMessage());
                }
            }, 10, 30, TimeUnit.SECONDS);
        }
        /* client.setProxy(new Proxy(Proxy.Type.HTTP, new InetSocketAddress("proxyaddress", 80)));*/
    } catch (InterruptedException e) {
        log.info("websocket connection...exception....", e);
    }
}

同步策略

如下图所示，soul-admin 在用户发生配置变更之后，会通过 EventPublisher 发出配置变更通知，由 EventDispatcher 处理该变更通知，然后根据配置的同步策略(http、weboscket、zookeeper)，将配置发送给对应的事件处理器

  • 如果是 websocket 同步策略，则将变更后的数据主动推送给 soul-web，并且在网关层，会有对应的 WebsocketCacheHandler 处理器处理来处 admin 的数据推送
  • 如果是 zookeeper 同步策略，将变更数据更新到 zookeeper，而 ZookeeperSyncCache 会监听到 zookeeper 的数据变更，并予以处理
  • 如果是 http 同步策略，soul-web 主动发起长轮询请求，默认有 90s 超时时间，如果 soul-admin 没有数据变更，则会阻塞 http 请求，如果有数据发生变更则响应变更的数据信息，如果超过 60s 仍然没有数据变更则响应空数据，网关层接到响应后，继续发起 http 请求，反复同样的请求

Link