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
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
| /**
* 增强版 Nacos 数据源（支持双向同步）
* 功能：
* 1. 从 Nacos 读取初始配置
* 2. 监听 Nacos 配置变更
* 3. 本地配置变更时自动写回 Nacos
*
* @param <T> 配置数据类型（如 List<FlowRule>）
*/
@Slf4j
public class NacosDataSource<T> extends AbstractDataSource<String, T> {
// 默认从Nacos获取配置的超时时间（毫秒）
private static final int DEFAULT_TIMEOUT = 3000;
/**
* 单线程线程池（用于处理配置变更通知）
* 设计说明：
* - 核心/最大线程数=1：确保配置变更顺序处理
* - 队列容量=1：防止积压过多变更请求
* - 拒绝策略=DiscardOldestPolicy：当队列满时丢弃最老任务
*/
private final ExecutorService pool = new ThreadPoolExecutor(
1, 1, 0, TimeUnit.MILLISECONDS,
new ArrayBlockingQueue<>(1),
new NamedThreadFactory("sentinel-nacos-ds-update", true),
new ThreadPoolExecutor.DiscardOldestPolicy()
);
// Nacos配置变更监听器
private final Listener configListener;
// Nacos配置的组ID（如 DEFAULT_GROUP）
private final String groupId;
// Nacos配置的数据ID（如 mianshimao-sentinel）
private final String dataId;
// 连接Nacos的配置属性
private final Properties properties;
// Nacos配置服务实例（注意：初始化失败可能为null）
private ConfigService configService = null;
// 是否启用自动发布到Nacos
private final boolean autoPublish;
// 双向转换器（支持对象与字符串互转）
private final BidirectionalConverter<String, T> bidirectionalConverter;
/**
* 构造方法（简化版，默认启用自动发布）
*/
public NacosDataSource(final String serverAddr, final String groupId, final String dataId,
BidirectionalConverter<String, T> converter) {
this(serverAddr, groupId, dataId, converter, true);
}
/**
* 构造方法（可控制自动发布）
*/
public NacosDataSource(final String serverAddr, final String groupId, final String dataId,
BidirectionalConverter<String, T> converter, boolean autoPublish) {
this(buildProperties(serverAddr), groupId, dataId, converter, autoPublish);
}
/**
* 核心构造方法
* @param properties Nacos连接配置
* @param groupId 配置组ID
* @param dataId 配置数据ID
* @param converter 双向转换器
* @param autoPublish 是否自动发布变更到Nacos
*/
public NacosDataSource(final Properties properties, final String groupId, final String dataId,
BidirectionalConverter<String, T> converter, boolean autoPublish) {
super(converter);
// 参数校验
if (StringUtil.isBlank(groupId) || StringUtil.isBlank(dataId)) {
throw new IllegalArgumentException("groupId和dataId不能为空");
}
AssertUtil.notNull(properties, "Nacos配置属性不能为null");
AssertUtil.notNull(converter, "转换器不能为null");
// 初始化字段
this.groupId = groupId;
this.dataId = dataId;
this.properties = properties;
this.autoPublish = autoPublish;
this.bidirectionalConverter = converter;
// 创建Nacos配置监听器
this.configListener = new Listener() {
@Override
public Executor getExecutor() {
return pool; // 使用专用线程池处理变更事件
}
@Override
public void receiveConfigInfo(final String configInfo) {
handleConfigUpdate(configInfo); // 处理配置变更
}
};
// 注册本地变更监听器（用于自动发布到Nacos）
if (autoPublish) {
getProperty().addListener(new PropertyListener<T>() {
@Override
public void configUpdate(T newValue) {
log.info("检测到本地配置变更，准备同步到Nacos...");
boolean success = publishConfig(newValue);
log.info("同步结果: {}", success ? "成功" : "失败");
}
@Override
public void configLoad(T value) {
// 初始加载时不触发发布
}
});
}
// 初始化Nacos连接和监听
initNacosListener();
// 加载初始配置
loadInitialConfig();
}
/**
* 处理Nacos配置变更
* @param configInfo 配置内容（JSON字符串）
*/
private void handleConfigUpdate(String configInfo) {
try {
log.info("接收到Nacos配置变更通知 dataId={}, groupId={}", dataId, groupId);
// 1. 转换配置格式
T newValue = bidirectionalConverter.convert(configInfo);
// 2. 更新本地属性（会自动触发监听器）
getProperty().updateValue(newValue);
} catch (Exception ex) {
log.error("配置转换失败 dataId=" + dataId, ex);
}
}
/**
* 发布配置到Nacos
* @param config 配置对象
* @return 是否发布成功
*/
public boolean publishConfig(T config) {
// 1. 检查自动发布是否启用
if (!autoPublish) {
log.warn("自动发布功能未启用");
return false;
}
// 2. 检查Nacos客户端是否初始化
if (configService == null) {
log.error("Nacos ConfigService未初始化");
return false;
}
try {
// 3. 转换配置为字符串
String content = bidirectionalConverter.reverseConvert(config);
log.debug("准备发布配置到Nacos:\n{}", content);
// 4. 调试信息输出
log.info("=== 发布配置调试信息 ===");
log.info("Nacos地址: {}", properties.getProperty(PropertyKeyConst.SERVER_ADDR));
log.info("命名空间: {}", properties.getProperty(PropertyKeyConst.NAMESPACE, "public"));
log.info("DataID: {}", dataId);
log.info("Group: {}", groupId);
// 5. 执行发布（指定内容类型为JSON）
boolean result = configService.publishConfig(dataId, groupId, content, "json");
if (result) {
log.info("配置发布成功 dataId={}", dataId);
} else {
log.error("Nacos返回发布失败");
}
return result;
} catch (NacosException e) {
log.error("Nacos发布配置失败 code={}, msg={}", e.getErrCode(), e.getMessage(), e);
return false;
}
}
/**
* 加载初始配置
*/
private void loadInitialConfig() {
try {
T newValue = loadConfig();
if (newValue == null) {
log.warn("初始配置为空 dataId={}", dataId);
}
getProperty().updateValue(newValue);
} catch (Exception ex) {
log.error("加载初始配置失败", ex);
}
}
/**
* 初始化Nacos监听器
*/
private void initNacosListener() {
try {
// 1. 创建ConfigService实例
this.configService = NacosFactory.createConfigService(this.properties);
log.info("Nacos ConfigService初始化成功");
// 2. 添加配置监听器
configService.addListener(dataId, groupId, configListener);
log.info("Nacos监听器注册成功 dataId={}", dataId);
} catch (Exception e) {
log.error("Nacos监听器初始化失败", e);
}
}
/**
* 从数据源读取配置（实现父类抽象方法）
*/
@Override
public String readSource() throws Exception {
if (configService == null) {
throw new IllegalStateException("Nacos ConfigService未初始化");
}
return configService.getConfig(dataId, groupId, DEFAULT_TIMEOUT);
}
/**
* 关闭资源
*/
@Override
public void close() {
// 1. 移除Nacos监听器
if (configService != null) {
configService.removeListener(dataId, groupId, configListener);
try {
configService.shutDown();
log.info("Nacos ConfigService已关闭");
} catch (Exception e) {
log.error("关闭Nacos ConfigService失败", e);
}
}
// 2. 关闭线程池
pool.shutdownNow();
log.info("Nacos数据源线程池已关闭");
}
/**
* 构建基础Nacos配置
*/
private static Properties buildProperties(String serverAddr) {
Properties properties = new Properties();
properties.setProperty(PropertyKeyConst.SERVER_ADDR, serverAddr);
return properties;
}
}
|
