零、前戏
最近在“玩”demo的过程中，就一直好奇。插件信息等元数据，soul是怎么从soul-admin同步到soul-bootstrap中的。
本着大胆瞎猜，小心求证的心态，就有今天的关于websocket这篇文章。
一、服务端--soul-admin
- 首先从jar入手，soul-admin中的只有一个spring-boot-starter-websocket官方jar包。
猜测：服务端实现是在写在soul-admin项目内部
- 通过全文搜索“websocket”，找到了DataSyncConfiguration
image
涉及websocket的配置初始化代码如下：
image
与yml中的配置项对应
image
- 进入WebsocketCollector中一探究竟
image
@ServerEndpoint("/websocket")科普帖子：https://blog.csdn.net/weixin_39793752/article/details/80949128
@ServerEndpoint("/websocket")
public class WebsocketCollector {
private static final Set<Session> SESSION_SET = new CopyOnWriteArraySet<>();
private static final String SESSION_KEY = "sessionKey";
/**
* On open.
*
* @param session the session
*/
@OnOpen
public void onOpen(final Session session) {
log.info("websocket on open successful....");
SESSION_SET.add(session);
}
/**
* On message.
*
* @param message the message
* @param session the session
*/
@OnMessage
public void onMessage(final String message, final Session session) {
if (message.equals(DataEventTypeEnum.MYSELF.name())) {
try {
ThreadLocalUtil.put(SESSION_KEY, session);
SpringBeanUtils.getInstance().getBean(SyncDataService.class).syncAll(DataEventTypeEnum.MYSELF);
} finally {
ThreadLocalUtil.clear();
}
}
}
/**
* On close.
*
* @param session the session
*/
@OnClose
public void onClose(final Session session) {
SESSION_SET.remove(session);
ThreadLocalUtil.clear();
}
/**
* On error.
*
* @param session the session
* @param error the error
*/
@OnError
public void onError(final Session session, final Throwable error) {
SESSION_SET.remove(session);
ThreadLocalUtil.clear();
log.error("websocket collection error: ", error);
}
/**
* Send.
*
* @param message the message
* @param type the type
*/
public static void send(final String message, final DataEventTypeEnum type) {
if (StringUtils.isNotBlank(message)) {
if (DataEventTypeEnum.MYSELF == type) {
try {
Session session = (Session) ThreadLocalUtil.get(SESSION_KEY);
if (session != null) {
session.getBasicRemote().sendText(message);
}
} catch (IOException e) {
log.error("websocket send result is exception: ", e);
}
return;
}
for (Session session : SESSION_SET) {
try {
session.getBasicRemote().sendText(message);
} catch (IOException e) {
log.error("websocket send result is exception: ", e);
}
}
}
}
}
@OnOpen：连接建立成功调用的方法
@OnClose：连接关闭调用方法
@OnMessage：收到客户端消息后调用的方法
@OnError：发送错误时调用
- 当服务端收到一条MYSELF类型的消息，将同步全量元数据信息
- admin 会推送一次全量数据，后续如果配置数据发生变更则将增量数据通过 websocket 主动推送给 soul-web
详细机制此处暂留坑，后文再续
image
二、客户端--soul-bootstrap
从pom文件入手，找到了soul-spring-boot-starter-sync-data-websocket中的WebsocketSyncDataConfiguration类
image
- 深入WebsocketSyncDataService
image
- 按照配置soul-admin的个数，创建线程池大小
- 通过线程池的定时任务，实现心跳机制----保持心跳
- /使用调度线程池进行断线重连，30秒进行一次。如果连接断开，将client.connectBlocking(3000, TimeUnit.MILLISECONDS);重连，重新发送MYSELF类型信息，进行全量更新。
image
线程池科普贴：https://blog.csdn.net/weixin_35756522/article/details/81707276
public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit);
command：执行线程
initialDelay：初始化延时
period：两次开始执行最小间隔时间
unit：计时单位
- SoulWebsocketClient 是具体处理websocket类
image
- 在与服务端建立连接时，会发送MYSELF类型的消息。****如上文所述，服务端接收****MYSELF类****型消息后将发送全量消息用于同步。
小结：
- soul-admin与soul-bootstrap基本通信链路，基本捋顺。相关数据处理的细节，将在后续文章中浅析。
- 日拱一卒，每天进步一点点
