You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@shenyu.apache.org by xi...@apache.org on 2021/09/14 06:08:33 UTC
[incubator-shenyu-website] branch main updated: add a blog about
websocket data sync source code analysis (#275)
This is an automated email from the ASF dual-hosted git repository.
xiaoyu pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-shenyu-website.git
The following commit(s) were added to refs/heads/main by this push:
new fc073dd add a blog about websocket data sync source code analysis (#275)
fc073dd is described below
commit fc073ddf91b3ba3131f54f08f456159fd5bbee7a
Author: midnight2104 <33...@qq.com>
AuthorDate: Tue Sep 14 14:08:30 2021 +0800
add a blog about websocket data sync source code analysis (#275)
* add websocket data sync source code analysis
* fix md lint
* fix md lint
---
...9-13-SourceCode-Analysis-WebSocket-Data-Sync.md | 1065 ++++++++++++++++++++
...9-13-SourceCode-Analysis-WebSocket-Data-Sync.md | 1061 +++++++++++++++++++
.../add-selector.png | Bin 0 -> 77766 bytes
.../data-changed-listener.png | Bin 0 -> 16959 bytes
.../data-handler.png | Bin 0 -> 17446 bytes
.../handler-selector.png | Bin 0 -> 123655 bytes
.../websocket-data-sync-admin-en.png | Bin 0 -> 58734 bytes
.../websocket-data-sync-admin.png | Bin 0 -> 58870 bytes
.../websocket-data-sync-gateway-en.png | Bin 0 -> 76134 bytes
.../websocket-data-sync-gateway.png | Bin 0 -> 78445 bytes
.../websocket-springboot-starter.png | Bin 0 -> 58525 bytes
11 files changed, 2126 insertions(+)
diff --git a/blog/2021-09-13-SourceCode-Analysis-WebSocket-Data-Sync.md b/blog/2021-09-13-SourceCode-Analysis-WebSocket-Data-Sync.md
new file mode 100644
index 0000000..4660e02
--- /dev/null
+++ b/blog/2021-09-13-SourceCode-Analysis-WebSocket-Data-Sync.md
@@ -0,0 +1,1065 @@
+---
+slug: code-analysis-websocket-data-sync
+title: WebSocket Data Synchronization Source Code Analysis
+author: Liu Liang
+author_title: Apache ShenYu Committer
+author_url: https://github.com/midnight2104
+tags: [websocket,data sync,Apache ShenYu]
+---
+
+
+In `ShenYu` gateway, data synchronization refers to how to synchronize the updated data to the gateway after the data is sent in the background management system. The Apache ShenYu gateway currently supports data synchronization for `ZooKeeper`, `WebSocket`, `http long poll`, `Nacos`, `etcd` and `Consul`. The main content of this article is based on `WebSocket` data synchronization source code analysis.
+
+
+> This paper based on `shenyu-2.4.0` version of the source code analysis, the official website of the introduction of please refer to the [Data Synchronization Design](https://shenyu.apache.org/docs/design/data-sync/) .
+
+
+### 1. About WebSocket Communication
+
+The WebSocket protocol was born in 2008 and became an international standard in 2011. It can be two-way communication, the server can take the initiative to push information to the client, the client can also take the initiative to send information to the server. The WebSocket protocol is based on the TCP protocol and belongs to the application layer, with low performance overhead and high communication efficiency. The protocol identifier is `ws`.
+
+
+### 2. Admin Data Sync
+
+Let's trace the source code from a real case, such as adding a selector data in the background management system:
+
+
+
+#### 2.1 Accept Changed Data
+
+- SelectorController.createSelector()
+
+Enter the createSelector() method of the `SelectorController` class, which validates data, adds or updates data, and returns results.
+
+
+```java
+@Validated
+@RequiredArgsConstructor
+@RestController
+@RequestMapping("/selector")
+public class SelectorController {
+
+ @PostMapping("")
+ public ShenyuAdminResult createSelector(@Valid @RequestBody final SelectorDTO selectorDTO) { // @Valid 数校验
+ // create or update data
+ Integer createCount = selectorService.createOrUpdate(selectorDTO);
+ // return result
+ return ShenyuAdminResult.success(ShenyuResultMessage.CREATE_SUCCESS, createCount);
+ }
+
+ // ......
+}
+```
+
+
+
+#### 2.2 Handle Data
+
+- SelectorServiceImpl.createOrUpdate()
+
+Convert data in the `SelectorServiceImpl` class using the `createOrUpdate()` method, save it to the database, publish the event, update `upstream`.
+
+
+```java
+@RequiredArgsConstructor
+@Service
+public class SelectorServiceImpl implements SelectorService {
+ // eventPublisher
+ private final ApplicationEventPublisher eventPublisher;
+
+ @Override
+ @Transactional(rollbackFor = Exception.class)
+ public int createOrUpdate(final SelectorDTO selectorDTO) {
+ int selectorCount;
+ // build data DTO --> DO
+ SelectorDO selectorDO = SelectorDO.buildSelectorDO(selectorDTO);
+ List<SelectorConditionDTO> selectorConditionDTOs = selectorDTO.getSelectorConditions();
+ // insert or update ?
+ if (StringUtils.isEmpty(selectorDTO.getId())) {
+ // insert into data
+ selectorCount = selectorMapper.insertSelective(selectorDO);
+ // insert into condition data
+ selectorConditionDTOs.forEach(selectorConditionDTO -> {
+ selectorConditionDTO.setSelectorId(selectorDO.getId());
+ selectorConditionMapper.insertSelective(SelectorConditionDO.buildSelectorConditionDO(selectorConditionDTO));
+ });
+ // check selector add
+ if (dataPermissionMapper.listByUserId(JwtUtils.getUserInfo().getUserId()).size() > 0) {
+ DataPermissionDTO dataPermissionDTO = new DataPermissionDTO();
+ dataPermissionDTO.setUserId(JwtUtils.getUserInfo().getUserId());
+ dataPermissionDTO.setDataId(selectorDO.getId());
+ dataPermissionDTO.setDataType(AdminConstants.SELECTOR_DATA_TYPE);
+ dataPermissionMapper.insertSelective(DataPermissionDO.buildPermissionDO(dataPermissionDTO));
+ }
+
+ } else {
+ // update data, delete and then insert
+ selectorCount = selectorMapper.updateSelective(selectorDO);
+ //delete rule condition then add
+ selectorConditionMapper.deleteByQuery(new SelectorConditionQuery(selectorDO.getId()));
+ selectorConditionDTOs.forEach(selectorConditionDTO -> {
+ selectorConditionDTO.setSelectorId(selectorDO.getId());
+ SelectorConditionDO selectorConditionDO = SelectorConditionDO.buildSelectorConditionDO(selectorConditionDTO);
+ selectorConditionMapper.insertSelective(selectorConditionDO);
+ });
+ }
+ // publish event
+ publishEvent(selectorDO, selectorConditionDTOs);
+
+ // update upstream
+ updateDivideUpstream(selectorDO);
+ return selectorCount;
+ }
+
+
+ // ......
+
+}
+
+```
+
+In the `Service` class to persist data, i.e. to the database, this should be familiar, not expand. The update upstream operation is analyzed in the corresponding section below, focusing on the publish event operation, which performs data synchronization.
+
+
+
+The logic of the `publishEvent()` method is to find the plugin corresponding to the selector, build the conditional data, and publish the change data.
+
+
+```java
+ private void publishEvent(final SelectorDO selectorDO, final List<SelectorConditionDTO> selectorConditionDTOs) {
+ // find plugin of selector
+ PluginDO pluginDO = pluginMapper.selectById(selectorDO.getPluginId());
+ // build condition data
+ List<ConditionData> conditionDataList = selectorConditionDTOs.stream().map(ConditionTransfer.INSTANCE::mapToSelectorDTO).collect(Collectors.toList());
+ // publish event
+ eventPublisher.publishEvent(new DataChangedEvent(ConfigGroupEnum.SELECTOR, DataEventTypeEnum.UPDATE,
+ Collections.singletonList(SelectorDO.transFrom(selectorDO, pluginDO.getName(), conditionDataList))));
+ }
+```
+
+Change data released by `eventPublisher.PublishEvent()` is complete, the `eventPublisher` object is a `ApplicationEventPublisher` class, The fully qualified class name is `org.springframework.context.ApplicationEventPublisher`. Here we see that publishing data is done through `Spring` related functionality.
+
+
+> `ApplicationEventPublisher`：
+>
+> When a state change, the publisher calls `ApplicationEventPublisher` of `publishEvent` method to release an event, `Spring` container broadcast event for all observers, The observer's `onApplicationEvent` method is called to pass the event object to the observer. There are two ways to call `publishEvent` method, one is to implement the interface by the container injection `ApplicationEventPublisher` object and then call the method, the other is a direct call container, the method of tw [...]
+>
+> - `ApplicationEventPublisher`: publish event;
+> - `ApplicationEvent`: `Spring` event, record the event source, time, and data;
+> - `ApplicationListener`: event listener, observer.
+
+In Spring event publishing mechanism, there are three objects,
+
+
+An object is a publish event `ApplicationEventPublisher`, in `ShenYu` through the constructor in the injected a `eventPublisher`.
+
+
+The other object is `ApplicationEvent` , inherited from `ShenYu` through `DataChangedEvent`, representing the event object.
+
+
+```java
+public class DataChangedEvent extends ApplicationEvent {
+//......
+}
+```
+
+The last object is `ApplicationListener` in `ShenYu` in through `DataChangedEventDispatcher` class implements this interface, as the event listener, responsible for handling the event object.
+
+
+```java
+@Component
+public class DataChangedEventDispatcher implements ApplicationListener<DataChangedEvent>, InitializingBean {
+
+ //......
+
+}
+```
+
+
+
+#### 2.3 Dispatch Data
+
+- DataChangedEventDispatcher.onApplicationEvent()
+
+Released when the event is completed, will automatically enter the `DataChangedEventDispatcher` class `onApplicationEvent()` method of handling events.
+
+
+```java
+@Component
+public class DataChangedEventDispatcher implements ApplicationListener<DataChangedEvent>, InitializingBean {
+
+ /**
+ * This method is called when there are data changes
+ * @param event
+ */
+ @Override
+ @SuppressWarnings("unchecked")
+ public void onApplicationEvent(final DataChangedEvent event) {
+ // Iterate through the data change listener (usually using a data synchronization approach is fine)
+ for (DataChangedListener listener : listeners) {
+ // What kind of data has changed
+ switch (event.getGroupKey()) {
+ case APP_AUTH: // app auth data
+ listener.onAppAuthChanged((List<AppAuthData>) event.getSource(), event.getEventType());
+ break;
+ case PLUGIN: // plugin data
+ listener.onPluginChanged((List<PluginData>) event.getSource(), event.getEventType());
+ break;
+ case RULE: // rule data
+ listener.onRuleChanged((List<RuleData>) event.getSource(), event.getEventType());
+ break;
+ case SELECTOR: // selector data
+ listener.onSelectorChanged((List<SelectorData>) event.getSource(), event.getEventType());
+ break;
+ case META_DATA: // metadata
+ listener.onMetaDataChanged((List<MetaData>) event.getSource(), event.getEventType());
+ break;
+ default: // Other types throw exception
+ throw new IllegalStateException("Unexpected value: " + event.getGroupKey());
+ }
+ }
+ }
+
+}
+```
+
+When there is a data change, the `onApplicationEvent` method is called and all the data change listeners are iterated to determine the data type and handed over to the appropriate data listener for processing.
+
+ShenYu groups all the data into five categories: APP_AUTH, PLUGIN, RULE, SELECTOR and META_DATA.
+
+Here the data change listener (`DataChangedListener`) is an abstraction of the data synchronization policy. Its concrete implementation is:
+
+
+
+
+These implementation classes are the synchronization strategies currently supported by ShenYu:
+
+- `WebsocketDataChangedListener`: data synchronization based on Websocket;
+- `ZookeeperDataChangedListener`:data synchronization based on Zookeeper;
+- `ConsulDataChangedListener`: data synchronization based on Consul;
+- `EtcdDataDataChangedListener`：data synchronization based on etcd;
+- `HttpLongPollingDataChangedListener`：data synchronization based on http long polling;
+- `NacosDataChangedListener`：data synchronization based on nacos;
+
+Given that there are so many implementation strategies, how do you decide which to use?
+
+Because this paper is based on `websocket` data synchronization source code analysis, so here to `WebsocketDataChangedListener` as an example, the analysis of how it is loaded and implemented.
+
+A global search in the source code project shows that its implementation is done in the `DataSyncConfiguration` class.
+
+```java
+/**
+ * Data Sync Configuration
+ * By springboot conditional assembly
+ * The type Data sync configuration.
+ */
+@Configuration
+public class DataSyncConfiguration {
+
+ /**
+ * The WebsocketListener(default strategy).
+ */
+ @Configuration
+ @ConditionalOnProperty(name = "shenyu.sync.websocket.enabled", havingValue = "true", matchIfMissing = true)
+ @EnableConfigurationProperties(WebsocketSyncProperties.class)
+ static class WebsocketListener {
+
+ /**
+ * Config event listener data changed listener.
+ * @return the data changed listener
+ */
+ @Bean
+ @ConditionalOnMissingBean(WebsocketDataChangedListener.class)
+ public DataChangedListener websocketDataChangedListener() {
+ return new WebsocketDataChangedListener();
+ }
+
+ /**
+ * Websocket collector.
+ * Websocket collector class: establish a connection, send a message, close the connection and other operations
+ * @return the websocket collector
+ */
+ @Bean
+ @ConditionalOnMissingBean(WebsocketCollector.class)
+ public WebsocketCollector websocketCollector() {
+ return new WebsocketCollector();
+ }
+
+ /**
+ * Server endpoint exporter
+ *
+ * @return the server endpoint exporter
+ */
+ @Bean
+ @ConditionalOnMissingBean(ServerEndpointExporter.class)
+ public ServerEndpointExporter serverEndpointExporter() {
+ return new ServerEndpointExporter();
+ }
+ }
+
+ //......
+}
+
+```
+
+This configuration class is implemented through the `SpringBoot` conditional assembly class. The `WebsocketListener` class has several annotations:
+
+
+- `@Configuration`: Configuration file, application context;
+
+- `@ConditionalOnProperty(name = "shenyu.sync.websocket.enabled", havingValue = "true", matchIfMissing = true)`: attribute condition. The configuration class takes effect only when the condition is met. That is, when we have the following configuration, `websocket` is used for data synchronization. Note, however, the `matchIfMissing = true` attribute, which means that this configuration class will work if you don't have the following configuration. Data synchronization based on `webSocke [...]
+
+ ```properties
+ shenyu:
+ sync:
+ websocket:
+ enabled: true
+ ```
+
+- `@EnableConfigurationProperties`：enable configuration properties;
+
+
+When we take the initiative to configuration, use the `websocket` data synchronization, `WebsocketDataChangedListener` is generated. So in the event handler `onApplicationEvent()`, it goes to the corresponding `listener`. In our case, a selector is to increase the new data, the data by adopting the `websocket`, so, the code will enter the `WebsocketDataChangedListener` selector data change process.
+
+
+```java
+ @Override
+ @SuppressWarnings("unchecked")
+ public void onApplicationEvent(final DataChangedEvent event) {
+ // Iterate through the data change listener (usually using a data synchronization approach is fine)
+ for (DataChangedListener listener : listeners) {
+ // What kind of data has changed
+ switch (event.getGroupKey()) {
+
+ // other logic is omitted
+ case SELECTOR: // selector data
+ listener.onSelectorChanged((List<SelectorData>) event.getSource(), event.getEventType()); // WebsocketDataChangedListener handle selector data
+ break;
+ }
+ }
+```
+
+
+
+#### 2.4 Websocket Data Changed Listener
+
+- WebsocketDataChangedListener.onSelectorChanged()
+
+In the `onSelectorChanged()` method, the data is encapsulated into `WebsocketData` and then sent via `webSocketCollector.send()`.
+
+
+```java
+ // selector data has been updated
+ @Override
+ public void onSelectorChanged(final List<SelectorData> selectorDataList, final DataEventTypeEnum eventType) {
+ // build WebsocketData
+ WebsocketData<SelectorData> websocketData =
+ new WebsocketData<>(ConfigGroupEnum.SELECTOR.name(), eventType.name(), selectorDataList);
+ // websocket send data
+ WebsocketCollector.send(GsonUtils.getInstance().toJson(websocketData), eventType);
+ }
+```
+
+
+
+#### 2.5 Websocket Send Data
+
+- WebsocketCollector.send()
+
+In the `send()` method, the type of synchronization is determined and processed according to the different types.
+
+```java
+@Slf4j
+@ServerEndpoint(value = "/websocket", configurator = WebsocketConfigurator.class)
+public class WebsocketCollector {
+
+/**
+ * Send.
+ *
+ * @param message the message
+ * @param type the type
+ */
+ public static void send(final String message, final DataEventTypeEnum type) {
+ if (StringUtils.isNotBlank(message)) {
+ // If it's MYSELF (first full synchronization)
+ if (DataEventTypeEnum.MYSELF == type) {
+ // get the session from ThreadLocal
+ Session session = (Session) ThreadLocalUtil.get(SESSION_KEY);
+ if (session != null) {
+ // send full data to the session
+ sendMessageBySession(session, message);
+ }
+ } else {
+ // subsequent incremental synchronization
+ // synchronize change data to all sessions
+ SESSION_SET.forEach(session -> sendMessageBySession(session, message));
+ }
+ }
+ }
+
+ private static void sendMessageBySession(final Session session, final String message) {
+ try {
+ // The message is sent through the Websocket session
+ session.getBasicRemote().sendText(message);
+ } catch (IOException e) {
+ log.error("websocket send result is exception: ", e);
+ }
+ }
+}
+```
+
+
+
+The example we give is a new operation, an incremental synchronization, so it goes
+
+`SESSION_SET.forEach(session -> sendMessageBySession(session, message));`
+
+
+then through
+
+`session.getBasicRemote().sendText(message);`
+
+the data was sent out.
+
+At this point, when data changes occur on the admin side, the changed data is increments sent to the gateway through the `WebSocket`.
+
+At this point, do you have any questions? For example, where does session come from? How does the gateway establish a connection with admin?
+
+Don't worry, let's do the synchronization analysis on the gateway side.
+
+However, before continuing with the source code analysis, let's use a diagram to string together the above analysis process.
+
+
+
+
+
+### 3. Gateway Data Sync
+
+Assume `ShenYu` gateway is already in normal operation, using the data synchronization mode is also `websocket`. How does the gateway receive and process new selector data from admin and send it to the gateway via WebSocket? Let's continue our source code analysis to find out.
+
+
+#### 3.1 WebsocketClient Accept Data
+
+- ShenyuWebsocketClient.onMessage()
+
+There is a `ShenyuWebsocketClient` class on the gateway, which inherits from `WebSocketClient` and can establish a connection and communicate with `WebSocket`.
+
+```java
+public final class ShenyuWebsocketClient extends WebSocketClient {
+ // ......
+}
+```
+
+After sending data via `websocket` on the admin side, `ShenyuWebsocketClient` can receive data via `onMessage()` and then process it itself.
+
+
+```java
+public final class ShenyuWebsocketClient extends WebSocketClient {
+ // execute after receiving the message
+ @Override
+ public void onMessage(final String result) {
+ // handle accept data
+ handleResult(result);
+ }
+
+ private void handleResult(final String result) {
+ // data deserialization
+ WebsocketData websocketData = GsonUtils.getInstance().fromJson(result, WebsocketData.class);
+ // which data types, plug-ins, selectors, rules...
+ ConfigGroupEnum groupEnum = ConfigGroupEnum.acquireByName(websocketData.getGroupType());
+ // which operation type, update, delete...
+ String eventType = websocketData.getEventType();
+ String json = GsonUtils.getInstance().toJson(websocketData.getData());
+
+ // handle data
+ websocketDataHandler.executor(groupEnum, json, eventType);
+ }
+}
+```
+
+After receiving the data, first has carried on the deserialization operation, read the data type and operation type, then hand over to `websocketDataHandler.executor()` for processing.
+
+#### 3.2 Execute Websocket Data Handler
+
+- WebsocketDataHandler.executor()
+
+A `Websocket` data handler is created in factory mode, providing one handler for each data type:
+
+
+> plugin --> PluginDataHandler;
+>
+> selector --> SelectorDataHandler;
+>
+> rule --> RuleDataHandler;
+>
+> auth --> AuthDataHandler;
+>
+> metadata --> MetaDataHandler.
+
+```java
+
+/**
+ * Create Websocket data handlers through factory mode
+ * The type Websocket cache handler.
+ */
+public class WebsocketDataHandler {
+
+ private static final EnumMap<ConfigGroupEnum, DataHandler> ENUM_MAP = new EnumMap<>(ConfigGroupEnum.class);
+
+ /**
+ * Instantiates a new Websocket data handler.
+ * @param pluginDataSubscriber the plugin data subscriber
+ * @param metaDataSubscribers the meta data subscribers
+ * @param authDataSubscribers the auth data subscribers
+ */
+ public WebsocketDataHandler(final PluginDataSubscriber pluginDataSubscriber,
+ final List<MetaDataSubscriber> metaDataSubscribers,
+ final List<AuthDataSubscriber> authDataSubscribers) {
+ // plugin --> PluginDataHandler
+ ENUM_MAP.put(ConfigGroupEnum.PLUGIN, new PluginDataHandler(pluginDataSubscriber));
+ // selector --> SelectorDataHandler
+ ENUM_MAP.put(ConfigGroupEnum.SELECTOR, new SelectorDataHandler(pluginDataSubscriber));
+ // rule --> RuleDataHandler
+ ENUM_MAP.put(ConfigGroupEnum.RULE, new RuleDataHandler(pluginDataSubscriber));
+ // auth --> AuthDataHandler
+ ENUM_MAP.put(ConfigGroupEnum.APP_AUTH, new AuthDataHandler(authDataSubscribers));
+ // metadata --> MetaDataHandler
+ ENUM_MAP.put(ConfigGroupEnum.META_DATA, new MetaDataHandler(metaDataSubscribers));
+ }
+
+ /**
+ * Executor.
+ *
+ * @param type the type
+ * @param json the json
+ * @param eventType the event type
+ */
+ public void executor(final ConfigGroupEnum type, final String json, final String eventType) {
+ // find the corresponding data handler based on the data type
+ ENUM_MAP.get(type).handle(json, eventType);
+ }
+}
+
+```
+
+Different data types have different ways of handling data, so there are different implementation classes. But they also have the same processing logic between them, so they can be implemented through the template approach to design patterns. The same logic is placed in the `handle()` method of the abstract class, and the different logic is handed over to the respective implementation class.
+
+
+
+
+In our case, a new selector is added, so it will be passed to the `SelectorDataHandler` for data processing.
+
+
+#### 3.3 Determine the Event Type
+
+- AbstractDataHandler.handle()
+
+Implement common logical handling of data changes: invoke different methods based on different operation types.
+
+```java
+
+public abstract class AbstractDataHandler<T> implements DataHandler {
+
+ /**
+ * Convert list.
+ * The different logic is implemented by the respective implementation classes
+ * @param json the json
+ * @return the list
+ */
+ protected abstract List<T> convert(String json);
+
+ /**
+ * Do refresh.
+ * The different logic is implemented by the respective implementation classes
+ * @param dataList the data list
+ */
+ protected abstract void doRefresh(List<T> dataList);
+
+ /**
+ * Do update.
+ * The different logic is implemented by the respective implementation classes
+ * @param dataList the data list
+ */
+ protected abstract void doUpdate(List<T> dataList);
+
+ /**
+ * Do delete.
+ * The different logic is implemented by the respective implementation classes
+ * @param dataList the data list
+ */
+ protected abstract void doDelete(List<T> dataList);
+
+ // General purpose logic, abstract class implementation
+ @Override
+ public void handle(final String json, final String eventType) {
+ List<T> dataList = convert(json);
+ if (CollectionUtils.isNotEmpty(dataList)) {
+ DataEventTypeEnum eventTypeEnum = DataEventTypeEnum.acquireByName(eventType);
+ switch (eventTypeEnum) {
+ case REFRESH:
+ case MYSELF:
+ doRefresh(dataList); //Refreshes data and synchronizes all data
+ break;
+ case UPDATE:
+ case CREATE:
+ doUpdate(dataList); // Update or create data, incremental synchronization
+ break;
+ case DELETE:
+ doDelete(dataList); // delete data
+ break;
+ default:
+ break;
+ }
+ }
+ }
+}
+```
+
+New selector data, new operation, through `switch-case` into `doUpdate()` method.
+
+#### 3.4 Enter the Specific Data Handler
+
+- SelectorDataHandler.doUpdate()
+
+```java
+
+/**
+ * The type Selector data handler.
+ */
+@RequiredArgsConstructor
+public class SelectorDataHandler extends AbstractDataHandler<SelectorData> {
+
+ private final PluginDataSubscriber pluginDataSubscriber;
+
+ //......
+
+ // update data
+ @Override
+ protected void doUpdate(final List<SelectorData> dataList) {
+ dataList.forEach(pluginDataSubscriber::onSelectorSubscribe);
+ }
+}
+```
+
+Iterate over the data and enter the `onSelectorSubscribe()` method.
+
+- PluginDataSubscriber.onSelectorSubscribe()
+
+It has no additional logic and calls the `subscribeDataHandler()` method directly. Within methods, there are data types (plugins, selectors, or rules) and action types (update or delete) to perform different logic.
+
+```java
+/**
+ * The universal plug-in data subscriber, responsible for handling all plug-in, selector, and rule information
+ * The type Common plugin data subscriber.
+ */
+public class CommonPluginDataSubscriber implements PluginDataSubscriber {
+ //......
+ // handle selector data
+ @Override
+ public void onSelectorSubscribe(final SelectoData selectorData) {
+ subscribeDataHandler(selectorData, DataEventTypeEnum.UPDATE);
+ }
+
+ // A subscription data handler that handles updates or deletions of data
+ private <T> void subscribeDataHandler(final T classData, final DataEventTypeEnum dataType) {
+ Optional.ofNullable(classData).ifPresent(data -> {
+ // plugin data
+ if (data instanceof PluginData) {
+ PluginData pluginData = (PluginData) data;
+ if (dataType == DataEventTypeEnum.UPDATE) { // update
+ // save the data to gateway memory
+ BaseDataCache.getInstance().cachePluginData(pluginData);
+ // If each plugin has its own processing logic, then do it
+ Optional.ofNullable(handlerMap.get(pluginData.getName())).ifPresent(handler -> handler.handlerPlugin(pluginData));
+ } else if (dataType == DataEventTypeEnum.DELETE) { // delete
+ // delete the data from gateway memory
+ BaseDataCache.getInstance().removePluginData(pluginData);
+ // If each plugin has its own processing logic, then do it
+ Optional.ofNullable(handlerMap.get(pluginData.getName())).ifPresent(handler -> handler.removePlugin(pluginData));
+ }
+ } else if (data instanceof SelectorData) { // selector data
+ SelectorData selectorData = (SelectorData) data;
+ if (dataType == DataEventTypeEnum.UPDATE) { // update
+ // save the data to gateway memory
+ BaseDataCache.getInstance().cacheSelectData(selectorData);
+ // If each plugin has its own processing logic, then do it
+ Optional.ofNullable(handlerMap.get(selectorData.getPluginName())).ifPresent(handler -> handler.handlerSelector(selectorData));
+ } else if (dataType == DataEventTypeEnum.DELETE) { // delete
+ // delete the data from gateway memory
+ BaseDataCache.getInstance().removeSelectData(selectorData);
+ // If each plugin has its own processing logic, then do it
+ Optional.ofNullable(handlerMap.get(selectorData.getPluginName())).ifPresent(handler -> handler.removeSelector(selectorData));
+ }
+ } else if (data instanceof RuleData) { // rule data
+ RuleData ruleData = (RuleData) data;
+ if (dataType == DataEventTypeEnum.UPDATE) { // update
+ // save the data to gateway memory
+ BaseDataCache.getInstance().cacheRuleData(ruleData);
+ // If each plugin has its own processing logic, then do it
+ Optional.ofNullable(handlerMap.get(ruleData.getPluginName())).ifPresent(handler -> handler.handlerRule(ruleData));
+ } else if (dataType == DataEventTypeEnum.DELETE) { // delete
+ // delete the data from gateway memory
+ BaseDataCache.getInstance().removeRuleData(ruleData);
+ // If each plugin has its own processing logic, then do it
+ Optional.ofNullable(handlerMap.get(ruleData.getPluginName())).ifPresent(handler -> handler.removeRule(ruleData));
+ }
+ }
+ });
+ }
+
+}
+```
+
+Adding a selector will enter the following logic:
+
+```java
+// save the data to gateway memory
+BaseDataCache.getInstance().cacheSelectData(selectorData);
+// If each plugin has its own processing logic, then do it
+Optional.ofNullable(handlerMap.get(selectorData.getPluginName())).ifPresent(handler -> handler.handlerSelector(selectorData));
+```
+
+One is to save the data to the gateway's memory. BaseDataCache is the class that ultimately caches data, implemented in a singleton pattern. The selector data is stored in the `SELECTOR_MAP` Map. In the subsequent use, also from this data.
+
+
+```java
+public final class BaseDataCache {
+ // private instance
+ private static final BaseDataCache INSTANCE = new BaseDataCache();
+ // private constructor
+ private BaseDataCache() {
+ }
+
+ /**
+ * Gets instance.
+ * public method
+ * @return the instance
+ */
+ public static BaseDataCache getInstance() {
+ return INSTANCE;
+ }
+
+ /**
+ * A Map of the cache selector data
+ * pluginName -> SelectorData.
+ */
+ private static final ConcurrentMap<String, List<SelectorData>> SELECTOR_MAP = Maps.newConcurrentMap();
+
+ public void cacheSelectData(final SelectorData selectorData) {
+ Optional.ofNullable(selectorData).ifPresent(this::selectorAccept);
+ }
+
+ /**
+ * cache selector data.
+ * @param data the selector data
+ */
+ private void selectorAccept(final SelectorData data) {
+ String key = data.getPluginName();
+ if (SELECTOR_MAP.containsKey(key)) { // Update operation, delete before insert
+ List<SelectorData> existList = SELECTOR_MAP.get(key);
+ final List<SelectorData> resultList = existList.stream().filter(r -> !r.getId().equals(data.getId())).collect(Collectors.toList());
+ resultList.add(data);
+ final List<SelectorData> collect = resultList.stream().sorted(Comparator.comparing(SelectorData::getSort)).collect(Collectors.toList());
+ SELECTOR_MAP.put(key, collect);
+ } else { // Add new operations directly to Map
+ SELECTOR_MAP.put(key, Lists.newArrayList(data));
+ }
+ }
+
+}
+```
+
+Second, if each plugin has its own processing logic, then do it. Through the `IDEA` editor, you can see that after adding a selector, there are the following plugins and processing. We're not going to expand it here.
+
+
+
+
+After the above source tracing, and through a practical case, in the 'admin' side to add a selector data, will `websocket` data synchronization process analysis cleared.
+
+Let's use the following figure to concatenate the data synchronization process on the gateway side:
+
+
+
+
+The data synchronization process has been analyzed, but there are still some problems that have not been analyzed, that is, how does the gateway establish a connection with admin?
+
+### 4. The Gateway Establishes a Websocket Connection with Admin
+
+- websocket config
+
+With the following configuration in the gateway configuration file and the dependency introduced, the `websocket` related service is started.
+
+
+```yaml
+shenyu:
+ file:
+ enabled: true
+ cross:
+ enabled: true
+ dubbo :
+ parameter: multi
+ sync:
+ websocket : # Use websocket for data synchronization
+ urls: ws://localhost:9095/websocket # websocket address of admin
+```
+
+Add a dependency on websocket in the gateway.
+
+
+```xml
+<!--shenyu data sync start use websocket-->
+<dependency>
+ <groupId>org.apache.shenyu</groupId>
+ <artifactId>shenyu-spring-boot-starter-sync-data-websocket</artifactId>
+ <version>${project.version}</version>
+</dependency>
+```
+
+- Websocket Data Sync Config
+
+The associated bean is created by conditional assembly of springboot. In the gateway started, if we configure the `shenyu.sync.websocket.urls`, then `websocket` data synchronization configuration will be loaded. The dependency loading is done through the `springboot starter`.
+
+
+```java
+
+/**
+ * WebsocketSyncDataService
+ * Conditional injection is implemented through SpringBoot
+ * Websocket sync data configuration for spring boot.
+ */
+@Configuration
+@ConditionalOnClass(WebsocketSyncDataService.class)
+@ConditionalOnProperty(prefix = "shenyu.sync.websocket", name = "urls")
+@Slf4j
+public class WebsocketSyncDataConfiguration {
+
+ /**
+ * Websocket sync data service.
+ * @param websocketConfig the websocket config
+ * @param pluginSubscriber the plugin subscriber
+ * @param metaSubscribers the meta subscribers
+ * @param authSubscribers the auth subscribers
+ * @return the sync data service
+ */
+ @Bean
+ public SyncDataService websocketSyncDataService(final ObjectProvider<WebsocketConfig> websocketConfig, final ObjectProvider<PluginDataSubscriber> pluginSubscriber,
+ final ObjectProvider<List<MetaDataSubscriber>> metaSubscribers, final ObjectProvider<List<AuthDataSubscriber>> authSubscribers) {
+ log.info("you use websocket sync shenyu data.......");
+ return new WebsocketSyncDataService(websocketConfig.getIfAvailable(WebsocketConfig::new), pluginSubscriber.getIfAvailable(),
+ metaSubscribers.getIfAvailable(Collections::emptyList), authSubscribers.getIfAvailable(Collections::emptyList));
+ }
+
+ /**
+ * Config websocket config.
+ *
+ * @return the websocket config
+ */
+ @Bean
+ @ConfigurationProperties(prefix = "shenyu.sync.websocket")
+ public WebsocketConfig websocketConfig() {
+ return new WebsocketConfig();
+ }
+}
+```
+
+Start a new `spring.factories` file in the `resources/META-INF` directory of your project and specify the configuration classes in the file.
+
+
+
+
+- WebsocketSyncDataService
+
+The following things are done in 'WebsocketSyncDataService' :
+
+- Read configuration `urls`, which represent the admin side of the synchronization address, if there are more than one, use "," split;
+
+- Create a scheduling thread pool, with each `admin` assigned one to perform scheduled tasks;
+
+- Create `ShenyuWebsocketClient`, assign one to each `admin`, set up `websocket` communication with `admin`;
+
+- Start connection with admin end `websocket`;
+
+- Executes a scheduled task every 10 seconds. The main function is to determine whether the `websocket` connection has been disconnected, if so, try to reconnect. If not, a `ping-pong` test is performed.
+
+```java
+
+/**
+ * Websocket sync data service.
+ */
+@Slf4j
+public class WebsocketSyncDataService implements SyncDataService, AutoCloseable {
+
+ private final List<WebSocketClient> clients = new ArrayList<>();
+
+ private final ScheduledThreadPoolExecutor executor;
+
+ /**
+ * Instantiates a new Websocket sync cache.
+ * @param websocketConfig the websocket config
+ * @param pluginDataSubscriber the plugin data subscriber
+ * @param metaDataSubscribers the meta data subscribers
+ * @param authDataSubscribers the auth data subscribers
+ */
+ public WebsocketSyncDataService(final WebsocketConfig websocketConfig,
+ final PluginDataSubscriber pluginDataSubscriber,
+ final List<MetaDataSubscriber> metaDataSubscribers,
+ final List<AuthDataSubscriber> authDataSubscribers) {
+ // If there are multiple synchronization addresses on the admin side, use commas (,) to separate them
+ String[] urls = StringUtils.split(websocketConfig.getUrls(), ",");
+ // Create a scheduling thread pool, one for each admin
+ executor = new ScheduledThreadPoolExecutor(urls.length, ShenyuThreadFactory.create("websocket-connect", true));
+ for (String url : urls) {
+ try {
+ //Create a WebsocketClient and assign one to each admin
+ clients.add(new ShenyuWebsocketClient(new URI(url), Objects.requireNonNull(pluginDataSubscriber), metaDataSubscribers, authDataSubscribers));
+ } catch (URISyntaxException e) {
+ log.error("websocket url({}) is error", url, e);
+ }
+ }
+ try {
+ for (WebSocketClient client : clients) {
+ // Establish a connection with the WebSocket Server
+ boolean success = client.connectBlocking(3000, TimeUnit.MILLISECONDS);
+ if (success) {
+ log.info("websocket connection is successful.....");
+ } else {
+ log.error("websocket connection is error.....");
+ }
+
+ // Run a scheduled task every 10 seconds
+ // The main function is to check whether the WebSocket connection is disconnected. If the connection is disconnected, retry the connection.
+ // If it is not disconnected, the ping-pong test is performed
+ executor.scheduleAtFixedRate(() -> {
+ try {
+ if (client.isClosed()) {
+ boolean reconnectSuccess = client.reconnectBlocking();
+ if (reconnectSuccess) {
+ log.info("websocket reconnect server[{}] is successful.....", client.getURI().toString());
+ } else {
+ log.error("websocket reconnection server[{}] is error.....", client.getURI().toString());
+ }
+ } else {
+ client.sendPing();
+ log.debug("websocket send to [{}] ping message successful", client.getURI().toString());
+ }
+ } catch (InterruptedException e) {
+ log.error("websocket connect is error :{}", e.getMessage());
+ }
+ }, 10, 10, TimeUnit.SECONDS);
+ }
+ /* client.setProxy(new Proxy(Proxy.Type.HTTP, new InetSocketAddress("proxyaddress", 80)));*/
+ } catch (InterruptedException e) {
+ log.info("websocket connection...exception....", e);
+ }
+
+ }
+
+ @Override
+ public void close() {
+ // close websocket client
+ for (WebSocketClient client : clients) {
+ if (!client.isClosed()) {
+ client.close();
+ }
+ }
+ // close threadpool
+ if (Objects.nonNull(executor)) {
+ executor.shutdown();
+ }
+ }
+}
+```
+
+- ShenyuWebsocketClient
+
+The `WebSocket` client created in `ShenYu` to communicate with the `admin` side. After the connection is successfully established for the first time, full data is synchronized and incremental data is subsequently synchronized.
+
+```java
+
+/**
+ * The type shenyu websocket client.
+ */
+@Slf4j
+public final class ShenyuWebsocketClient extends WebSocketClient {
+
+ private volatile boolean alreadySync = Boolean.FALSE;
+
+ private final WebsocketDataHandler websocketDataHandler;
+
+ /**
+ * Instantiates a new shenyu websocket client.
+ * @param serverUri the server uri
+ * @param pluginDataSubscriber the plugin data subscriber
+ * @param metaDataSubscribers the meta data subscribers
+ * @param authDataSubscribers the auth data subscribers
+ */
+ public ShenyuWebsocketClient(final URI serverUri, final PluginDataSubscriber pluginDataSubscriber,final List<MetaDataSubscriber> metaDataSubscribers, final List<AuthDataSubscriber> authDataSubscribers) {
+ super(serverUri);
+ this.websocketDataHandler = new WebsocketDataHandler(pluginDataSubscriber, metaDataSubscribers, authDataSubscribers);
+ }
+
+ // Execute after the connection is successfully established
+ @Override
+ public void onOpen(final ServerHandshake serverHandshake) {
+ // To prevent re-execution when reconnecting, use alreadySync to determine
+ if (!alreadySync) {
+ // Synchronize all data, type MYSELF
+ send(DataEventTypeEnum.MYSELF.name());
+ alreadySync = true;
+ }
+ }
+
+ // Execute after receiving the message
+ @Override
+ public void onMessage(final String result) {
+ // handle data
+ handleResult(result);
+ }
+
+ // Execute after shutdown
+ @Override
+ public void onClose(final int i, final String s, final boolean b) {
+ this.close();
+ }
+
+ // Execute after error
+ @Override
+ public void onError(final Exception e) {
+ this.close();
+ }
+
+ @SuppressWarnings("ALL")
+ private void handleResult(final String result) {
+ // Data deserialization
+ WebsocketData websocketData = GsonUtils.getInstance().fromJson(result, WebsocketData.class);
+ // Which data types, plugins, selectors, rules...
+ ConfigGroupEnum groupEnum = ConfigGroupEnum.acquireByName(websocketData.getGroupType());
+ // Which operation type, update, delete...
+ String eventType = websocketData.getEventType();
+ String json = GsonUtils.getInstance().toJson(websocketData.getData());
+
+ // handle data
+ websocketDataHandler.executor(groupEnum, json, eventType);
+ }
+}
+
+```
+
+### 5. Summary
+
+This paper through a practical case, the data synchronization principle of websocket source code analysis. The main knowledge points involved are as follows:
+
+- `WebSocket` supports bidirectional communication and has good performance. It is recommended.
+
+- Complete event publishing and listening via `Spring`;
+
+- Support multiple synchronization strategies through abstract `DataChangedListener` interface, interface oriented programming;
+
+- Use factory mode to create `WebsocketDataHandler` to handle different data types;
+
+- Implement `AbstractDataHandler` using template method design patterns to handle general operation types;
+
+- Use singleton design pattern to cache data class `BaseDataCache`;
+
+- Loading of configuration classes via conditional assembly of SpringBoot and starter loading mechanism.
+
diff --git a/i18n/zh/docusaurus-plugin-content-blog/2021-09-13-SourceCode-Analysis-WebSocket-Data-Sync.md b/i18n/zh/docusaurus-plugin-content-blog/2021-09-13-SourceCode-Analysis-WebSocket-Data-Sync.md
new file mode 100644
index 0000000..845d7a7
--- /dev/null
+++ b/i18n/zh/docusaurus-plugin-content-blog/2021-09-13-SourceCode-Analysis-WebSocket-Data-Sync.md
@@ -0,0 +1,1061 @@
+---
+slug: code-analysis-websocket-data-sync
+title: WebSocket数据同步源码分析
+author: Liu Liang
+author_title: Apache ShenYu Committer
+author_url: https://github.com/midnight2104
+tags: [websocket,data sync,Apache ShenYu]
+---
+
+
+在`ShenYu`网关中，数据同步是指，当在后台管理系统中，数据发送了更新后，如何将更新的数据同步到网关中。`Apache ShenYu` 网关当前支持`ZooKeeper`、`WebSocket`、`Http长轮询`、`Nacos` 、`etcd` 和 `Consul` 进行数据同步。本文的主要内容是基于`WebSocket`的数据同步源码分析。
+
+> 本文基于`shenyu-2.4.0`版本进行源码分析，官网的介绍请参考 [数据同步原理](https://shenyu.apache.org/zh/docs/design/data-sync) 。
+
+
+### 1. 关于WebSocket通信
+
+`WebSocket`协议诞生于`2008`年，在`2011`年成为国际标准。它可以双向通信，服务器可以主动向客户端推送信息，客户端也可以主动向服务器发送信息。`WebSocket`协议建立在 `TCP` 协议之上，属于应用层，性能开销小，通信高效，协议标识符是`ws`。
+
+
+### 2. Admin数据同步
+
+我们从一个实际案例进行源码追踪，比如在后台管理系统中，新增一条选择器数据：
+
+
+
+#### 2.1 接收数据
+
+- SelectorController.createSelector()
+
+进入`SelectorController`类中的`createSelector()`方法，它负责数据的校验，添加或更新数据，返回结果信息。
+
+```java
+@Validated
+@RequiredArgsConstructor
+@RestController
+@RequestMapping("/selector")
+public class SelectorController {
+
+ @PostMapping("")
+ public ShenyuAdminResult createSelector(@Valid @RequestBody final SelectorDTO selectorDTO) { // @Valid 数校验
+ // 添加或更新数据
+ Integer createCount = selectorService.createOrUpdate(selectorDTO);
+ // 返回结果信息
+ return ShenyuAdminResult.success(ShenyuResultMessage.CREATE_SUCCESS, createCount);
+ }
+
+ // ......
+}
+```
+
+
+
+#### 2.2 处理数据
+
+- SelectorServiceImpl.createOrUpdate()
+
+在`SelectorServiceImpl`类中通过`createOrUpdate()`方法完成数据的转换，保存到数据库，发布事件，更新`upstream`。
+
+```java
+@RequiredArgsConstructor
+@Service
+public class SelectorServiceImpl implements SelectorService {
+ // 负责事件发布的eventPublisher
+ private final ApplicationEventPublisher eventPublisher;
+
+ @Override
+ @Transactional(rollbackFor = Exception.class)
+ public int createOrUpdate(final SelectorDTO selectorDTO) {
+ int selectorCount;
+ // 构建数据 DTO --> DO
+ SelectorDO selectorDO = SelectorDO.buildSelectorDO(selectorDTO);
+ List<SelectorConditionDTO> selectorConditionDTOs = selectorDTO.getSelectorConditions();
+ // 判断是添加还是更新
+ if (StringUtils.isEmpty(selectorDTO.getId())) {
+ // 插入选择器数据
+ selectorCount = selectorMapper.insertSelective(selectorDO);
+ // 插入选择器中的条件数据
+ selectorConditionDTOs.forEach(selectorConditionDTO -> {
+ selectorConditionDTO.setSelectorId(selectorDO.getId());
+ selectorConditionMapper.insertSelective(SelectorConditionDO.buildSelectorConditionDO(selectorConditionDTO));
+ });
+ // check selector add
+ // 权限检查
+ if (dataPermissionMapper.listByUserId(JwtUtils.getUserInfo().getUserId()).size() > 0) {
+ DataPermissionDTO dataPermissionDTO = new DataPermissionDTO();
+ dataPermissionDTO.setUserId(JwtUtils.getUserInfo().getUserId());
+ dataPermissionDTO.setDataId(selectorDO.getId());
+ dataPermissionDTO.setDataType(AdminConstants.SELECTOR_DATA_TYPE);
+ dataPermissionMapper.insertSelective(DataPermissionDO.buildPermissionDO(dataPermissionDTO));
+ }
+
+ } else {
+ // 更新数据，先删除再新增
+ selectorCount = selectorMapper.updateSelective(selectorDO);
+ //delete rule condition then add
+ selectorConditionMapper.deleteByQuery(new SelectorConditionQuery(selectorDO.getId()));
+ selectorConditionDTOs.forEach(selectorConditionDTO -> {
+ selectorConditionDTO.setSelectorId(selectorDO.getId());
+ SelectorConditionDO selectorConditionDO = SelectorConditionDO.buildSelectorConditionDO(selectorConditionDTO);
+ selectorConditionMapper.insertSelective(selectorConditionDO);
+ });
+ }
+ // 发布事件
+ publishEvent(selectorDO, selectorConditionDTOs);
+
+ // 更新upstream
+ updateDivideUpstream(selectorDO);
+ return selectorCount;
+ }
+
+
+ // ......
+
+}
+
+```
+
+在`Service`类完成数据的持久化操作，即保存数据到数据库，这个大家应该很熟悉了，就不展开。关于更新`upstream`操作，放到后面对应的章节中进行分析，重点关注发布事件的操作，它会进行数据同步。
+
+
+
+`publishEvent()`方法的逻辑是：找到选择器对应的插件，构建条件数据，发布变更数据。
+
+```java
+ private void publishEvent(final SelectorDO selectorDO, final List<SelectorConditionDTO> selectorConditionDTOs) {
+ // 找到选择器对应的插件
+ PluginDO pluginDO = pluginMapper.selectById(selectorDO.getPluginId());
+ // 构建条件数据
+ List<ConditionData> conditionDataList = selectorConditionDTOs.stream().map(ConditionTransfer.INSTANCE::mapToSelectorDTO).collect(Collectors.toList());
+ // 发布变更数据
+ eventPublisher.publishEvent(new DataChangedEvent(ConfigGroupEnum.SELECTOR, DataEventTypeEnum.UPDATE,
+ Collections.singletonList(SelectorDO.transFrom(selectorDO, pluginDO.getName(), conditionDataList))));
+ }
+```
+
+发布变更数据通过`eventPublisher.publishEvent()`完成，这个`eventPublisher`对象是一个`ApplicationEventPublisher`类，这个类的全限定名是`org.springframework.context.ApplicationEventPublisher`。看到这儿，我们知道了发布数据是通过`Spring`相关的功能来完成的。
+
+> 关于`ApplicationEventPublisher`：
+>
+> 当有状态发生变化时，发布者调用 `ApplicationEventPublisher` 的 `publishEvent` 方法发布一个事件，`Spring`容器广播事件给所有观察者，调用观察者的 `onApplicationEvent` 方法把事件对象传递给观察者。调用 `publishEvent`方法有两种途径，一种是实现接口由容器注入 `ApplicationEventPublisher` 对象然后调用其方法，另一种是直接调用容器的方法，两种方法发布事件没有太大区别。
+>
+> - `ApplicationEventPublisher`：发布事件；
+> - `ApplicationEvent`：`Spring` 事件，记录事件源、时间和数据；
+> - `ApplicationListener`：事件监听者，观察者。
+
+在`Spring`的事件发布机制中，有三个对象，
+
+一个是发布事件的`ApplicationEventPublisher`，在`ShenYu`中通过构造器注入了一个`eventPublisher`。
+
+另一个对象是`ApplicationEvent`，在`ShenYu`中通过`DataChangedEvent`继承了它，表示事件对象。
+
+```java
+public class DataChangedEvent extends ApplicationEvent {
+//......
+}
+```
+
+最后一个是 `ApplicationListener`，在`ShenYu`中通过`DataChangedEventDispatcher`类实现了该接口，作为事件的监听者，负责处理事件对象。
+
+```java
+@Component
+public class DataChangedEventDispatcher implements ApplicationListener<DataChangedEvent>, InitializingBean {
+
+ //......
+
+}
+```
+
+
+
+#### 2.3 分发数据
+
+- DataChangedEventDispatcher.onApplicationEvent()
+
+当事件发布完成后，会自动进入到`DataChangedEventDispatcher`类中的`onApplicationEvent()`方法，进行事件处理。
+
+```java
+@Component
+public class DataChangedEventDispatcher implements ApplicationListener<DataChangedEvent>, InitializingBean {
+
+ /**
+ * 有数据变更时，调用此方法
+ * @param event
+ */
+ @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());
+ }
+ }
+ }
+
+}
+```
+
+当有数据变更时，调用`onApplicationEvent`方法，然后遍历所有数据变更监听器，判断是哪种数据类型，交给相应的数据监听器进行处理。
+
+`ShenYu`将所有数据进行了分组，一共是五种：认证信息、插件信息、规则信息、选择器信息和元数据。
+
+这里的数据变更监听器（`DataChangedListener`），就是数据同步策略的抽象，它的具体实现有：
+
+
+
+这几个实现类就是当前`ShenYu`支持的同步策略：
+
+- `WebsocketDataChangedListener`：基于`websocket`的数据同步；
+- `ZookeeperDataChangedListener`：基于`zookeeper`的数据同步；
+- `ConsulDataChangedListener`：基于`consul`的数据同步；
+- `EtcdDataDataChangedListener`：基于`etcd`的数据同步；
+- `HttpLongPollingDataChangedListener`：基于`http长轮询`的数据同步；
+- `NacosDataChangedListener`：基于`nacos`的数据同步；
+
+既然有这么多种实现策略，那么如何确定使用哪一种呢？
+
+因为本文是基于`websocket`的数据同步源码分析，所以这里以`WebsocketDataChangedListener`为例，分析它是如何被加载并实现的。
+
+通过在源码工程中进行全局搜索，可以看到，它的实现是在`DataSyncConfiguration`类完成的。
+
+```java
+/**
+ * 数据同步配置类
+ * 通过springboot条件装配实现
+ * The type Data sync configuration.
+ */
+@Configuration
+public class DataSyncConfiguration {
+
+ /**
+ * websocket数据同步（默认策略）
+ * The WebsocketListener(default strategy).
+ */
+ @Configuration
+ @ConditionalOnProperty(name = "shenyu.sync.websocket.enabled", havingValue = "true", matchIfMissing = true)
+ @EnableConfigurationProperties(WebsocketSyncProperties.class)
+ static class WebsocketListener {
+
+ /**
+ * Config event listener data changed listener.
+ * 配置websocket数据变更监听器
+ * @return the data changed listener
+ */
+ @Bean
+ @ConditionalOnMissingBean(WebsocketDataChangedListener.class)
+ public DataChangedListener websocketDataChangedListener() {
+ return new WebsocketDataChangedListener();
+ }
+
+ /**
+ * Websocket collector.
+ * Websocket处理类：建立连接，发送消息，关闭连接等操作
+ * @return the websocket collector
+ */
+ @Bean
+ @ConditionalOnMissingBean(WebsocketCollector.class)
+ public WebsocketCollector websocketCollector() {
+ return new WebsocketCollector();
+ }
+
+ /**
+ * Server endpoint exporter
+ *
+ * @return the server endpoint exporter
+ */
+ @Bean
+ @ConditionalOnMissingBean(ServerEndpointExporter.class)
+ public ServerEndpointExporter serverEndpointExporter() {
+ return new ServerEndpointExporter();
+ }
+ }
+
+ //......
+}
+
+```
+
+这个配置类是通过`SpringBoot`条件装配类实现的。在`WebsocketListener`类上面有几个注解：
+
+- `@Configuration`：配置文件，应用上下文；
+
+- `@ConditionalOnProperty(name = "shenyu.sync.websocket.enabled", havingValue = "true", matchIfMissing = true)`：属性条件判断，满足条件，该配置类才会生效。也就是说，当我们有如下配置时，就会采用`websocket`进行数据同步。不过，这里需要注意下`matchIfMissing = true`这个属性，它表示，如果你没有如下的配置，该配置类也会生效。基于`websocket`的数据同步时官方推荐的方式，也是默认采用的方式。
+
+ ```properties
+ shenyu:
+ sync:
+ websocket:
+ enabled: true
+ ```
+
+- `@EnableConfigurationProperties`：启用配置属性；
+
+
+
+当我们主动配置，采用`websocket`进行数据同步时，`WebsocketDataChangedListener`就会生成。所以在事件处理方法`onApplicationEvent()`中，就会到相应的`listener`中。在我们的案例中，是新增加了一条选择器数据，数据通过采用的是`websocket`，所以，代码会进入到`WebsocketDataChangedListener`进行选择器数据变更处理。
+
+```java
+ @Override
+ @SuppressWarnings("unchecked")
+ public void onApplicationEvent(final DataChangedEvent event) {
+ // 遍历数据变更监听器(一般使用一种数据同步的方式就好了)
+ for (DataChangedListener listener : listeners) {
+ // 哪种数据发生变更
+ switch (event.getGroupKey()) {
+
+ // 省略了其他逻辑
+
+ case SELECTOR: // 选择器信息
+ listener.onSelectorChanged((List<SelectorData>) event.getSource(), event.getEventType()); // WebsocketDataChangedListener进行选择器数据变更处理
+ break;
+ }
+ }
+```
+
+
+#### 2.4 Websocket数据变更监听器
+
+- WebsocketDataChangedListener.onSelectorChanged()
+
+ 在`onSelectorChanged()`方法中，将数据进行了封装，转成`WebsocketData`，然后通过`WebsocketCollector.send()`发送数据。
+
+```java
+ // 选择器数据有更新
+ @Override
+ public void onSelectorChanged(final List<SelectorData> selectorDataList, final DataEventTypeEnum eventType) {
+ // 构造 WebsocketData 数据
+ WebsocketData<SelectorData> websocketData =
+ new WebsocketData<>(ConfigGroupEnum.SELECTOR.name(), eventType.name(), selectorDataList);
+ // 通过websocket发送数据
+ WebsocketCollector.send(GsonUtils.getInstance().toJson(websocketData), eventType);
+ }
+```
+
+
+#### 2.5 Websocket发送数据
+
+- WebsocketCollector.send()
+
+在`send()`方法中，判断了一下同步的类型，根据不同的类型，进行处理。
+
+```java
+@Slf4j
+@ServerEndpoint(value = "/websocket", configurator = WebsocketConfigurator.class)
+public class WebsocketCollector {
+
+/**
+ * Send.
+ *
+ * @param message the message
+ * @param type the type
+ */
+ public static void send(final String message, final DataEventTypeEnum type) {
+ if (StringUtils.isNotBlank(message)) {
+ // 如果是MYSELF（第一次的全量同步）
+ if (DataEventTypeEnum.MYSELF == type) {
+ // 从threadlocal中获取session
+ Session session = (Session) ThreadLocalUtil.get(SESSION_KEY);
+ if (session != null) {
+ // 向该session发送全量数据
+ sendMessageBySession(session, message);
+ }
+ } else {
+ // 后续的增量同步
+ // 向所有的session中同步变更数据
+ SESSION_SET.forEach(session -> sendMessageBySession(session, message));
+ }
+ }
+ }
+
+ private static void sendMessageBySession(final Session session, final String message) {
+ try {
+ // 通过websocket的session把消息发送出去
+ session.getBasicRemote().sendText(message);
+ } catch (IOException e) {
+ log.error("websocket send result is exception: ", e);
+ }
+ }
+}
+```
+
+
+
+我们给的案例是一个新增操作 ，是一个增量同步，所以会走
+
+`SESSION_SET.forEach(session -> sendMessageBySession(session, message));`
+
+这个逻辑。
+
+再通过
+
+`session.getBasicRemote().sendText(message);`
+
+将数据发送了出去。
+
+
+
+至此，当`admin`端发生数据变更时，就将变更的数据以增量形式通过`WebSocket`发给了网关。
+
+分析到这里，不知道大家有没有疑问呢？比如`session`是怎么来的？网关如何和`admin`建立连接的？
+
+不要着急，我们接下来就进行网关端的同步分析。
+
+不过，在继续源码分析前，我们用一张图将上面的分析过程串联起来。
+
+
+
+
+
+### 3. 网关数据同步
+
+假设`ShenYu`网关已经在正常运行了，使用的数据同步方式也是`websocket`。那么当在`admin`端新增一条选择器数据后，并且通过`WebSocket`发送到网关，那网关是如何接收并处理数据的呢？接下来我们就继续进行源码分析，一探究竟。
+
+#### 3.1 WebsocketClient接收数据
+
+- ShenyuWebsocketClient.onMessage()
+
+在网关端有一个`ShenyuWebsocketClient`类，它继承了`WebSocketClient`，可以和`WebSocket`建立连接并通信。
+
+```java
+public final class ShenyuWebsocketClient extends WebSocketClient {
+ // ......
+}
+```
+
+当在`admin`端通过`websocket`发送数据后，`ShenyuWebsocketClient`就可以通过`onMessage()`接收到数据，然后就可以自己进行处理。
+
+```java
+public final class ShenyuWebsocketClient extends WebSocketClient {
+ // 接受到消息后执行
+ @Override
+ public void onMessage(final String result) {
+ // 处理接收到的数据
+ handleResult(result);
+ }
+
+ private void handleResult(final String result) {
+ // 数据反序列化
+ WebsocketData websocketData = GsonUtils.getInstance().fromJson(result, WebsocketData.class);
+ // 哪种数据类型，插件、选择器、规则...
+ ConfigGroupEnum groupEnum = ConfigGroupEnum.acquireByName(websocketData.getGroupType());
+ // 哪种操作类型，更新、删除...
+ String eventType = websocketData.getEventType();
+ String json = GsonUtils.getInstance().toJson(websocketData.getData());
+
+ // 处理数据
+ websocketDataHandler.executor(groupEnum, json, eventType);
+ }
+}
+```
+
+
+
+接收到数据后，首先进行了反序列化操作，读取数据类型和操作类型，紧接着，就交给`websocketDataHandler.executor()`进行处理。
+
+#### 3.2 执行Websocket事件处理器
+
+- WebsocketDataHandler.executor()
+
+通过工厂模式创建了`Websocket`数据处理器，每种数据类型，都提供了一个处理器：
+
+> 插件 --> 插件数据处理器;
+>
+> 选择器 --> 选择器数据处理器；
+>
+> 规则 --> 规则数据处理器；
+>
+> 认证信息 --> 认证数据处理器；
+>
+> 元数据 --> 元数据处理器。
+
+```java
+
+/**
+ * 通过工厂模式创建 Websocket数据处理器
+ * The type Websocket cache handler.
+ */
+public class WebsocketDataHandler {
+
+ private static final EnumMap<ConfigGroupEnum, DataHandler> ENUM_MAP = new EnumMap<>(ConfigGroupEnum.class);
+
+ /**
+ * Instantiates a new Websocket data handler.
+ * 每种数据类型，提供一个处理器
+ * @param pluginDataSubscriber the plugin data subscriber
+ * @param metaDataSubscribers the meta data subscribers
+ * @param authDataSubscribers the auth data subscribers
+ */
+ public WebsocketDataHandler(final PluginDataSubscriber pluginDataSubscriber,
+ final List<MetaDataSubscriber> metaDataSubscribers,
+ final List<AuthDataSubscriber> authDataSubscribers) {
+ // 插件 --> 插件数据处理器
+ ENUM_MAP.put(ConfigGroupEnum.PLUGIN, new PluginDataHandler(pluginDataSubscriber));
+ // 选择器 --> 选择器数据处理器
+ ENUM_MAP.put(ConfigGroupEnum.SELECTOR, new SelectorDataHandler(pluginDataSubscriber));
+ // 规则 --> 规则数据处理器
+ ENUM_MAP.put(ConfigGroupEnum.RULE, new RuleDataHandler(pluginDataSubscriber));
+ // 认证信息 --> 认证数据处理器
+ ENUM_MAP.put(ConfigGroupEnum.APP_AUTH, new AuthDataHandler(authDataSubscribers));
+ // 元数据 --> 元数据处理器
+ ENUM_MAP.put(ConfigGroupEnum.META_DATA, new MetaDataHandler(metaDataSubscribers));
+ }
+
+ /**
+ * Executor.
+ *
+ * @param type the type
+ * @param json the json
+ * @param eventType the event type
+ */
+ public void executor(final ConfigGroupEnum type, final String json, final String eventType) {
+ // 根据数据类型，找到对应的数据处理器
+ ENUM_MAP.get(type).handle(json, eventType);
+ }
+}
+
+```
+
+
+
+不同的数据类型，有不同的数据处理方式，所以有不同的实现类。但是它们之间也有相同的处理逻辑，所以可以通过模板方法设计模式来实现。相同的逻辑放在抽象类中的`handle()`方法中，不同逻辑就交给各自的实现类。
+
+
+
+
+
+我们的案例是新增了一条选择器数据，所以会交给`SelectorDataHandler`（ 选择器 --> 选择器数据处理器）进行数据处理。
+
+
+
+#### 3.3 判断事件类型
+
+- AbstractDataHandler.handle()
+
+实现数据变更的通用逻辑处理：根据不同的操作类型调用不同方法。
+
+```java
+
+public abstract class AbstractDataHandler<T> implements DataHandler {
+
+ /**
+ * Convert list.
+ * 不同的逻辑由各自实现类去实现
+ * @param json the json
+ * @return the list
+ */
+ protected abstract List<T> convert(String json);
+
+ /**
+ * Do refresh.
+ * 不同的逻辑由各自实现类去实现
+ * @param dataList the data list
+ */
+ protected abstract void doRefresh(List<T> dataList);
+
+ /**
+ * Do update.
+ * 不同的逻辑由各自实现类去实现
+ * @param dataList the data list
+ */
+ protected abstract void doUpdate(List<T> dataList);
+
+ /**
+ * Do delete.
+ * 不同的逻辑由各自实现类去实现
+ * @param dataList the data list
+ */
+ protected abstract void doDelete(List<T> dataList);
+
+ // 通用逻辑，抽象类实现
+ @Override
+ public void handle(final String json, final String eventType) {
+ List<T> dataList = convert(json);
+ if (CollectionUtils.isNotEmpty(dataList)) {
+ DataEventTypeEnum eventTypeEnum = DataEventTypeEnum.acquireByName(eventType);
+ switch (eventTypeEnum) {
+ case REFRESH:
+ case MYSELF:
+ doRefresh(dataList); //刷新数据，全量同步
+ break;
+ case UPDATE:
+ case CREATE:
+ doUpdate(dataList); // 更新或创建数据，增量同步
+ break;
+ case DELETE:
+ doDelete(dataList); // 删除数据
+ break;
+ default:
+ break;
+ }
+ }
+ }
+}
+```
+
+新增一条选择器数据，是新增操作，通过`switch-case`进入到`doUpdate()`方法中。
+
+#### 3.4 进入具体的数据处理器
+
+- SelectorDataHandler.doUpdate()
+
+```java
+
+/**
+ * 选择器数据处理器
+ * The type Selector data handler.
+ */
+@RequiredArgsConstructor
+public class SelectorDataHandler extends AbstractDataHandler<SelectorData> {
+
+ private final PluginDataSubscriber pluginDataSubscriber;
+
+ //......
+
+ // 更新操作
+ @Override
+ protected void doUpdate(final List<SelectorData> dataList) {
+ dataList.forEach(pluginDataSubscriber::onSelectorSubscribe);
+ }
+}
+```
+
+遍历数据，进入`onSelectorSubscribe()`方法。
+
+- PluginDataSubscriber.onSelectorSubscribe()
+
+它没有其他逻辑，直接调用`subscribeDataHandler()`方法。在方法中，更具数据类型（插件、选择器或规则），操作类型（更新或删除），去执行不同逻辑。
+
+```java
+/**
+ * 通用插件数据订阅者，负责处理所有插件、选择器和规则信息
+ * The type Common plugin data subscriber.
+ */
+public class CommonPluginDataSubscriber implements PluginDataSubscriber {
+ //......
+ // 处理选择器数据
+ @Override
+ public void onSelectorSubscribe(final SelectorData selectorData) {
+ subscribeDataHandler(selectorData, DataEventTypeEnum.UPDATE);
+ }
+
+ // 订阅数据处理器，处理数据的更新或删除
+ private <T> void subscribeDataHandler(final T classData, final DataEventTypeEnum dataType) {
+ Optional.ofNullable(classData).ifPresent(data -> {
+ // 插件数据
+ if (data instanceof PluginData) {
+ PluginData pluginData = (PluginData) data;
+ if (dataType == DataEventTypeEnum.UPDATE) { // 更新操作
+ // 将数据保存到网关内存
+ BaseDataCache.getInstance().cachePluginData(pluginData);
+ // 如果每个插件还有自己的处理逻辑，那么就去处理
+ Optional.ofNullable(handlerMap.get(pluginData.getName())).ifPresent(handler -> handler.handlerPlugin(pluginData));
+ } else if (dataType == DataEventTypeEnum.DELETE) { // 删除操作
+ // 从网关内存移除数据
+ BaseDataCache.getInstance().removePluginData(pluginData);
+ // 如果每个插件还有自己的处理逻辑，那么就去处理
+ Optional.ofNullable(handlerMap.get(pluginData.getName())).ifPresent(handler -> handler.removePlugin(pluginData));
+ }
+ } else if (data instanceof SelectorData) { // 选择器数据
+ SelectorData selectorData = (SelectorData) data;
+ if (dataType == DataEventTypeEnum.UPDATE) { // 更新操作
+ // 将数据保存到网关内存
+ BaseDataCache.getInstance().cacheSelectData(selectorData);
+ // 如果每个插件还有自己的处理逻辑，那么就去处理
+ Optional.ofNullable(handlerMap.get(selectorData.getPluginName())).ifPresent(handler -> handler.handlerSelector(selectorData));
+ } else if (dataType == DataEventTypeEnum.DELETE) { // 删除操作
+ // 从网关内存移除数据
+ BaseDataCache.getInstance().removeSelectData(selectorData);
+ // 如果每个插件还有自己的处理逻辑，那么就去处理
+ Optional.ofNullable(handlerMap.get(selectorData.getPluginName())).ifPresent(handler -> handler.removeSelector(selectorData));
+ }
+ } else if (data instanceof RuleData) { // 规则数据
+ RuleData ruleData = (RuleData) data;
+ if (dataType == DataEventTypeEnum.UPDATE) { // 更新操作
+ // 将数据保存到网关内存
+ BaseDataCache.getInstance().cacheRuleData(ruleData);
+ // 如果每个插件还有自己的处理逻辑，那么就去处理
+ Optional.ofNullable(handlerMap.get(ruleData.getPluginName())).ifPresent(handler -> handler.handlerRule(ruleData));
+ } else if (dataType == DataEventTypeEnum.DELETE) { // 删除操作
+ // 从网关内存移除数据
+ BaseDataCache.getInstance().removeRuleData(ruleData);
+ // 如果每个插件还有自己的处理逻辑，那么就去处理
+ Optional.ofNullable(handlerMap.get(ruleData.getPluginName())).ifPresent(handler -> handler.removeRule(ruleData));
+ }
+ }
+ });
+ }
+
+}
+```
+
+
+
+那么新增一条选择器数据，会进入下面的逻辑：
+
+```java
+// 将数据保存到网关内存
+BaseDataCache.getInstance().cacheSelectData(selectorData);
+// 如果每个插件还有自己的处理逻辑，那么就去处理
+Optional.ofNullable(handlerMap.get(selectorData.getPluginName())).ifPresent(handler -> handler.handlerSelector(selectorData));
+```
+
+一是将数据保存到网关的内存中。`BaseDataCache`是最终缓存数据的类，通过单例模式实现。选择器数据就存到了`SELECTOR_MAP`这个`Map`中。在后续使用的时候，也是从这里拿数据。
+
+```java
+public final class BaseDataCache {
+ // 私有变量
+ private static final BaseDataCache INSTANCE = new BaseDataCache();
+ // 私有构造器
+ private BaseDataCache() {
+ }
+
+ /**
+ * Gets instance.
+ * 公开方法
+ * @return the instance
+ */
+ public static BaseDataCache getInstance() {
+ return INSTANCE;
+ }
+
+ /**
+ * 缓存选择器数据的Map
+ * pluginName -> SelectorData.
+ */
+ private static final ConcurrentMap<String, List<SelectorData>> SELECTOR_MAP = Maps.newConcurrentMap();
+
+ public void cacheSelectData(final SelectorData selectorData) {
+ Optional.ofNullable(selectorData).ifPresent(this::selectorAccept);
+ }
+
+ /**
+ * cache selector data.
+ * 缓存选择器数据
+ * @param data the selector data
+ */
+ private void selectorAccept(final SelectorData data) {
+ String key = data.getPluginName();
+ if (SELECTOR_MAP.containsKey(key)) { // 更新操作，先删除再插入
+ List<SelectorData> existList = SELECTOR_MAP.get(key);
+ final List<SelectorData> resultList = existList.stream().filter(r -> !r.getId().equals(data.getId())).collect(Collectors.toList());
+ resultList.add(data);
+ final List<SelectorData> collect = resultList.stream().sorted(Comparator.comparing(SelectorData::getSort)).collect(Collectors.toList());
+ SELECTOR_MAP.put(key, collect);
+ } else { // 新增操作，直接放到Map中
+ SELECTOR_MAP.put(key, Lists.newArrayList(data));
+ }
+ }
+
+}
+```
+
+
+
+二是如果每个插件还有自己的处理逻辑，那么就去处理。 通过`idea`编辑器可以看到，当新增一条选择器后，有如下的插件还有处理。这里我们就不再展开了。
+
+
+
+经过以上的源码追踪，并通过一个实际的案例，在`admin`端新增一条选择器数据，就将`websocket`数据同步的流程分析清除了。
+
+我们还是用下面的一张图将网关端的数据同步流程串联一下：
+
+
+
+数据同步的流程已经分析完了，但是还有一些问题没有分析到，就是网关是如何跟`admin`建立连接的？
+
+### 4. 网关和admin建立websocket连接
+
+- websocket配置
+
+在网关的配置文件中有如下配置，并且引入了相关依赖，就会启动`websocket`相关服务。
+
+```yaml
+shenyu:
+ file:
+ enabled: true
+ cross:
+ enabled: true
+ dubbo :
+ parameter: multi
+ sync:
+ websocket : # 使用websocket进行数据同步
+ urls: ws://localhost:9095/websocket # admin端的websocket地址
+```
+
+在网关中引入`websocket`的依赖。
+
+```xml
+<!--shenyu data sync start use websocket-->
+<dependency>
+ <groupId>org.apache.shenyu</groupId>
+ <artifactId>shenyu-spring-boot-starter-sync-data-websocket</artifactId>
+ <version>${project.version}</version>
+</dependency>
+```
+
+- Websocket数据同步配置
+
+通过`springboot`的条件装配，创建相关的`bean`。在网关启动的时候，如果我们配置了`shenyu.sync.websocket.urls`，那么`Websocket`数据同步配置就会被加载。这里通过`spring boot starter`完成依赖的加载。
+
+```java
+
+/**
+ * Websocket数据同步配置
+ * 通过springboot实现条件注入
+ * Websocket sync data configuration for spring boot.
+ */
+@Configuration
+@ConditionalOnClass(WebsocketSyncDataService.class)
+@ConditionalOnProperty(prefix = "shenyu.sync.websocket", name = "urls")
+@Slf4j
+public class WebsocketSyncDataConfiguration {
+
+ /**
+ * Websocket sync data service.
+ * Websocket数据同步服务
+ * @param websocketConfig the websocket config
+ * @param pluginSubscriber the plugin subscriber
+ * @param metaSubscribers the meta subscribers
+ * @param authSubscribers the auth subscribers
+ * @return the sync data service
+ */
+ // 创建websocketSyncDataService
+ @Bean
+ public SyncDataService websocketSyncDataService(final ObjectProvider<WebsocketConfig> websocketConfig, final ObjectProvider<PluginDataSubscriber> pluginSubscriber,
+ final ObjectProvider<List<MetaDataSubscriber>> metaSubscribers, final ObjectProvider<List<AuthDataSubscriber>> authSubscribers) {
+ log.info("you use websocket sync shenyu data.......");
+ return new WebsocketSyncDataService(websocketConfig.getIfAvailable(WebsocketConfig::new), pluginSubscriber.getIfAvailable(),
+ metaSubscribers.getIfAvailable(Collections::emptyList), authSubscribers.getIfAvailable(Collections::emptyList));
+ }
+
+ /**
+ * Config websocket config.
+ *
+ * @return the websocket config
+ */
+ @Bean
+ @ConfigurationProperties(prefix = "shenyu.sync.websocket")
+ public WebsocketConfig websocketConfig() {
+ return new WebsocketConfig(); // 创建WebsocketConfig
+ }
+}
+```
+
+
+
+在项目的`resources/META-INF`目录先新建`spring.factories`文件，在文件中指明配置类。
+
+
+
+
+
+- Websocket数据同步服务
+
+在`WebsocketSyncDataService`中做了如下几件事情：
+
+- 读取配置中的`urls`，这个表示`admin`端的同步地址，有多个的话，使用","分割；
+- 创建调度线程池，一个`admin`分配一个，用于执行定时任务；
+- 创建`ShenyuWebsocketClient`，一个`admin`分配一个，用于和`admin`建立`websocket`通信；
+- 开始和`admin`端的`websocket` 建立连接；
+- 执行定时任务，每隔10秒执行一次。主要作用是判断`websocket`连接是否已经断开，如果已经断开，则尝试重连。如果没有断开，就进行 `ping-pong` 检测。
+
+```java
+
+/**
+ * Websocket数据同步服务
+ * Websocket sync data service.
+ */
+@Slf4j
+public class WebsocketSyncDataService implements SyncDataService, AutoCloseable {
+
+ private final List<WebSocketClient> clients = new ArrayList<>();
+
+ private final ScheduledThreadPoolExecutor executor;
+
+ /**
+ * Instantiates a new Websocket sync cache.
+ * 创建Websocket数据同步服务
+ * @param websocketConfig the websocket config
+ * @param pluginDataSubscriber the plugin data subscriber
+ * @param metaDataSubscribers the meta data subscribers
+ * @param authDataSubscribers the auth data subscribers
+ */
+ public WebsocketSyncDataService(final WebsocketConfig websocketConfig,
+ final PluginDataSubscriber pluginDataSubscriber,
+ final List<MetaDataSubscriber> metaDataSubscribers,
+ final List<AuthDataSubscriber> authDataSubscribers) {
+ // admin端的同步地址，有多个的话，使用","分割
+ String[] urls = StringUtils.split(websocketConfig.getUrls(), ",");
+ // 创建调度线程池，一个admin分配一个
+ executor = new ScheduledThreadPoolExecutor(urls.length, ShenyuThreadFactory.create("websocket-connect", true));
+ for (String url : urls) {
+ try {
+ //创建WebsocketClient，一个admin分配一个
+ clients.add(new ShenyuWebsocketClient(new URI(url), Objects.requireNonNull(pluginDataSubscriber), metaDataSubscribers, authDataSubscribers));
+ } catch (URISyntaxException e) {
+ log.error("websocket url({}) is error", url, e);
+ }
+ }
+ try {
+ for (WebSocketClient client : clients) {
+ // 和websocket server建立连接
+ boolean success = client.connectBlocking(3000, TimeUnit.MILLISECONDS);
+ if (success) {
+ log.info("websocket connection is successful.....");
+ } else {
+ log.error("websocket connection is error.....");
+ }
+
+ // 执行定时任务，每隔10秒执行一次
+ // 主要作用是判断websocket连接是否已经断开，如果已经断开，则尝试重连。
+ // 如果没有断开，就进行 ping-pong 检测
+ executor.scheduleAtFixedRate(() -> {
+ try {
+ if (client.isClosed()) {
+ boolean reconnectSuccess = client.reconnectBlocking();
+ if (reconnectSuccess) {
+ log.info("websocket reconnect server[{}] is successful.....", client.getURI().toString());
+ } else {
+ log.error("websocket reconnection server[{}] is error.....", client.getURI().toString());
+ }
+ } else {
+ client.sendPing();
+ log.debug("websocket send to [{}] ping message successful", client.getURI().toString());
+ }
+ } catch (InterruptedException e) {
+ log.error("websocket connect is error :{}", e.getMessage());
+ }
+ }, 10, 10, TimeUnit.SECONDS);
+ }
+ /* client.setProxy(new Proxy(Proxy.Type.HTTP, new InetSocketAddress("proxyaddress", 80)));*/
+ } catch (InterruptedException e) {
+ log.info("websocket connection...exception....", e);
+ }
+
+ }
+
+ @Override
+ public void close() {
+ // 关闭 websocket client
+ for (WebSocketClient client : clients) {
+ if (!client.isClosed()) {
+ client.close();
+ }
+ }
+ // 关闭线程池
+ if (Objects.nonNull(executor)) {
+ executor.shutdown();
+ }
+ }
+}
+```
+
+- ShenyuWebsocketClient
+
+在`ShenYu`中创建的`WebSocket`客户端，用于和`admin`端通信。第一次成功建立连接后，同步全量数据，后续进行增量同步。
+
+```java
+
+/**
+ * 在ShenYu中自定义的WebSocket客户端
+ * The type shenyu websocket client.
+ */
+@Slf4j
+public final class ShenyuWebsocketClient extends WebSocketClient {
+
+ private volatile boolean alreadySync = Boolean.FALSE;
+
+ private final WebsocketDataHandler websocketDataHandler;
+
+ /**
+ * Instantiates a new shenyu websocket client.
+ * 创建ShenyuWebsocketClient
+ * @param serverUri the server uri 服务端uri
+ * @param pluginDataSubscriber the plugin data subscriber 插件数据订阅器
+ * @param metaDataSubscribers the meta data subscribers 元数据订阅器
+ * @param authDataSubscribers the auth data subscribers 认证数据订阅器
+ */
+ public ShenyuWebsocketClient(final URI serverUri, final PluginDataSubscriber pluginDataSubscriber,final List<MetaDataSubscriber> metaDataSubscribers, final List<AuthDataSubscriber> authDataSubscribers) {
+ super(serverUri);
+ this.websocketDataHandler = new WebsocketDataHandler(pluginDataSubscriber, metaDataSubscribers, authDataSubscribers);
+ }
+
+ // 成功建立连接后执行
+ @Override
+ public void onOpen(final ServerHandshake serverHandshake) {
+ // 防止重新建立连接时，再次执行，所以用alreadySync进行判断
+ if (!alreadySync) {
+ // 同步所有数据，MYSELF 类型
+ send(DataEventTypeEnum.MYSELF.name());
+ alreadySync = true;
+ }
+ }
+
+ // 接收到消息后执行
+ @Override
+ public void onMessage(final String result) {
+ // 处理接收到的数据
+ handleResult(result);
+ }
+
+ // 关闭后执行
+ @Override
+ public void onClose(final int i, final String s, final boolean b) {
+ this.close();
+ }
+
+ // 失败后执行
+ @Override
+ public void onError(final Exception e) {
+ this.close();
+ }
+
+ @SuppressWarnings("ALL")
+ private void handleResult(final String result) {
+ // 数据反序列化
+ WebsocketData websocketData = GsonUtils.getInstance().fromJson(result, WebsocketData.class);
+ // 哪种数据类型，插件、选择器、规则...
+ ConfigGroupEnum groupEnum = ConfigGroupEnum.acquireByName(websocketData.getGroupType());
+ // 哪种操作类型，更新、删除...
+ String eventType = websocketData.getEventType();
+ String json = GsonUtils.getInstance().toJson(websocketData.getData());
+
+ // 处理数据
+ websocketDataHandler.executor(groupEnum, json, eventType);
+ }
+}
+
+```
+
+
+### 5. 总结
+
+本文通过一个实际案例，对`websocket`的数据同步原理进行了源码分析。涉及到的主要知识点如下：
+
+- `websocket`支持双向通信，性能好，推荐使用；
+- 通过`Spring`完成事件发布和监听；
+- 通过抽象`DataChangedListener`接口，支持多种同步策略，面向接口编程；
+- 使用工厂模式创建 `WebsocketDataHandler`，实现不同数据类型的处理；
+- 使用模板方法设计模式实现`AbstractDataHandler`，处理通用的操作类型；
+- 使用单例设计模式实现缓存数据类`BaseDataCache`；
+- 通过`SpringBoot`的条件装配和`starter`加载机制实现配置类的加载。
+
diff --git a/static/img/activities/code-analysis-websocket-data-sync/add-selector.png b/static/img/activities/code-analysis-websocket-data-sync/add-selector.png
new file mode 100644
index 0000000..67c24be
Binary files /dev/null and b/static/img/activities/code-analysis-websocket-data-sync/add-selector.png differ
diff --git a/static/img/activities/code-analysis-websocket-data-sync/data-changed-listener.png b/static/img/activities/code-analysis-websocket-data-sync/data-changed-listener.png
new file mode 100644
index 0000000..6a0e9d2
Binary files /dev/null and b/static/img/activities/code-analysis-websocket-data-sync/data-changed-listener.png differ
diff --git a/static/img/activities/code-analysis-websocket-data-sync/data-handler.png b/static/img/activities/code-analysis-websocket-data-sync/data-handler.png
new file mode 100644
index 0000000..811d5fc
Binary files /dev/null and b/static/img/activities/code-analysis-websocket-data-sync/data-handler.png differ
diff --git a/static/img/activities/code-analysis-websocket-data-sync/handler-selector.png b/static/img/activities/code-analysis-websocket-data-sync/handler-selector.png
new file mode 100644
index 0000000..2ff9d2b
Binary files /dev/null and b/static/img/activities/code-analysis-websocket-data-sync/handler-selector.png differ
diff --git a/static/img/activities/code-analysis-websocket-data-sync/websocket-data-sync-admin-en.png b/static/img/activities/code-analysis-websocket-data-sync/websocket-data-sync-admin-en.png
new file mode 100644
index 0000000..a27963e
Binary files /dev/null and b/static/img/activities/code-analysis-websocket-data-sync/websocket-data-sync-admin-en.png differ
diff --git a/static/img/activities/code-analysis-websocket-data-sync/websocket-data-sync-admin.png b/static/img/activities/code-analysis-websocket-data-sync/websocket-data-sync-admin.png
new file mode 100644
index 0000000..de6dfab
Binary files /dev/null and b/static/img/activities/code-analysis-websocket-data-sync/websocket-data-sync-admin.png differ
diff --git a/static/img/activities/code-analysis-websocket-data-sync/websocket-data-sync-gateway-en.png b/static/img/activities/code-analysis-websocket-data-sync/websocket-data-sync-gateway-en.png
new file mode 100644
index 0000000..0937406
Binary files /dev/null and b/static/img/activities/code-analysis-websocket-data-sync/websocket-data-sync-gateway-en.png differ
diff --git a/static/img/activities/code-analysis-websocket-data-sync/websocket-data-sync-gateway.png b/static/img/activities/code-analysis-websocket-data-sync/websocket-data-sync-gateway.png
new file mode 100644
index 0000000..c809aca
Binary files /dev/null and b/static/img/activities/code-analysis-websocket-data-sync/websocket-data-sync-gateway.png differ
diff --git a/static/img/activities/code-analysis-websocket-data-sync/websocket-springboot-starter.png b/static/img/activities/code-analysis-websocket-data-sync/websocket-springboot-starter.png
new file mode 100644
index 0000000..90ea10e
Binary files /dev/null and b/static/img/activities/code-analysis-websocket-data-sync/websocket-springboot-starter.png differ