lxy/redkale
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

优化DataSource配置中心通知

Browse Source
This commit is contained in:
Redkale
2022-12-07 22:21:47 +08:00
parent 1972e6308c
commit b41f34cade
2 changed files with 42 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files

95
src/main/java/org/redkale/boot/Application.java
View File

@@ -1772,13 +1772,13 @@ public final class Application {
void updateSourceProperties(Properties sourceChangeCache) { void updateSourceProperties(Properties sourceChangeCache) {
if (sourceChangeCache == null || sourceChangeCache.isEmpty()) return; if (sourceChangeCache == null || sourceChangeCache.isEmpty()) return;
synchronized (sourceProperties) { synchronized (sourceProperties) {
boolean same = true; Properties changedProps = new Properties();
for (Map.Entry<Object, Object> en : sourceChangeCache.entrySet()) { for (Map.Entry<Object, Object> en : sourceChangeCache.entrySet()) {
String key = en.getKey().toString(); String key = en.getKey().toString();
if (key.startsWith("redkale.datasource.") || key.startsWith("redkale.datasource[") if (key.startsWith("redkale.datasource.") || key.startsWith("redkale.datasource[")
|| key.startsWith("redkale.cachesource.") || key.startsWith("redkale.cachesource[")) { || key.startsWith("redkale.cachesource.") || key.startsWith("redkale.cachesource[")) {
if (!Objects.equals(en.getValue(), sourceProperties.get(key))) { if (!Objects.equals(en.getValue(), sourceProperties.get(key))) {
same = false; changedProps.put(en.getKey(), en.getValue());
if (key.endsWith(".name")) { //不更改source.name属性 if (key.endsWith(".name")) { //不更改source.name属性
throw new RuntimeException("source properties contains illegal key: " + key); throw new RuntimeException("source properties contains illegal key: " + key);
} }
@@ -1787,70 +1787,53 @@ public final class Application {
throw new RuntimeException("source properties contains illegal key: " + key); throw new RuntimeException("source properties contains illegal key: " + key);
} }
} }
if (same) return; //无内容改变 if (changedProps.isEmpty()) return; //无内容改变
AnyValue redNode = AnyValue.loadFromProperties(sourceChangeCache).getAnyValue("redkale"); AnyValue newRedNode = AnyValue.loadFromProperties(changedProps).getAnyValue("redkale");
AnyValue cacheNode = redNode.getAnyValue("cachesource"); AnyValue newCacheNode = newRedNode.getAnyValue("cachesource");
Map<String, String> back = new HashMap<>(); if (newCacheNode != null) {
if (cacheNode != null) { newCacheNode.forEach(null, (sourceName, newConf) -> {
cacheNode.forEach(null, (key, conf) -> { CacheSource source = Utility.find(cacheSources, s -> Objects.equals(s.resourceName(), sourceName));
CacheSource source = Utility.find(cacheSources, s -> Objects.equals(s.resourceName(), key)); if (source == null) return; //多余的数据源
if (source == null) return; DefaultAnyValue old = (DefaultAnyValue) findSourceConfig(sourceName, "cachesource");
back.clear(); old.merge(newConf);
List<ResourceEvent> events = new ArrayList<>(); List<ResourceEvent> events = new ArrayList<>();
DefaultAnyValue old = (DefaultAnyValue) findSourceConfig(key, "cachesource"); changedProps.forEach((k, v) -> {
conf.forEach((k, v) -> { final String key = k.toString();
if (old != null) { String prefix = "redkale.cachesource[" + sourceName + "].";
String o = old.getValue(k); int pos = key.indexOf(prefix);
back.put(k, o); if (pos < 0) {
events.add(ResourceEvent.create(k, v, o)); prefix = "redkale.cachesource." + sourceName + ".";
old.setValue(k, v); pos = key.indexOf(prefix);
} else {
events.add(ResourceEvent.create(k, v, null));
} }
if (pos < 0) return;
events.add(ResourceEvent.create(key.substring(prefix.length()), v, sourceProperties.get(key)));
}); });
try { ((AbstractCacheSource) source).onResourceChange(events.toArray(new ResourceEvent[events.size()]));
((AbstractCacheSource) source).onResourceChange(events.toArray(new ResourceEvent[events.size()]));
} catch (RuntimeException e) {
if (old != null) { //回退
back.forEach((k, v) -> {
old.setValue(k, v);
});
}
throw e;
}
}); });
} }
AnyValue sourceNode = redNode.getAnyValue("datasource"); AnyValue newSourceNode = newRedNode.getAnyValue("datasource");
if (sourceNode != null) { if (newSourceNode != null) {
sourceNode.forEach(null, (key, conf) -> { newSourceNode.forEach(null, (sourceName, newConf) -> {
DataSource source = Utility.find(dataSources, s -> Objects.equals(s.resourceName(), key)); DataSource source = Utility.find(dataSources, s -> Objects.equals(s.resourceName(), sourceName));
if (source == null) return; if (source == null) return; //多余的数据源
back.clear(); DefaultAnyValue old = (DefaultAnyValue) findSourceConfig(sourceName, "datasource");
old.merge(newConf);
List<ResourceEvent> events = new ArrayList<>(); List<ResourceEvent> events = new ArrayList<>();
DefaultAnyValue old = (DefaultAnyValue) findSourceConfig(key, "datasource"); changedProps.forEach((k, v) -> {
conf.forEach((k, v) -> { final String key = k.toString();
if (old != null) { String prefix = "redkale.datasource[" + sourceName + "].";
String o = old.getValue(k); int pos = key.indexOf(prefix);
back.put(k, o); if (pos < 0) {
events.add(ResourceEvent.create(k, v, o)); prefix = "redkale.datasource." + sourceName + ".";
old.setValue(k, v); pos = key.indexOf(prefix);
} else {
events.add(ResourceEvent.create(k, v, null));
} }
if (pos < 0) return;
events.add(ResourceEvent.create(key.substring(prefix.length()), v, sourceProperties.get(key)));
}); });
try { ((AbstractDataSource) source).onResourceChange(events.toArray(new ResourceEvent[events.size()]));
((AbstractDataSource) source).onResourceChange(events.toArray(new ResourceEvent[events.size()]));
} catch (RuntimeException e) {
if (old != null) { //回退
back.forEach((k, v) -> {
old.setValue(k, v);
});
}
throw e;
}
}); });
} }
sourceProperties.putAll(sourceChangeCache); sourceProperties.putAll(changedProps);
} }
} }

3
src/main/java/org/redkale/source/DataSqlSource.java
View File

@@ -81,11 +81,14 @@ public abstract class DataSqlSource extends AbstractDataSource implements Functi
//用于复制表结构使用 //用于复制表结构使用
protected String tablecopySQL; protected String tablecopySQL;
protected AnyValue config;
public DataSqlSource() { public DataSqlSource() {
} }
@Override @Override
public void init(AnyValue conf) { public void init(AnyValue conf) {
this.config = conf;
if (conf.getAnyValue("read") == null) { //没有读写分离 if (conf.getAnyValue("read") == null) { //没有读写分离
Properties rwConf = new Properties(); Properties rwConf = new Properties();
conf.forEach((k, v) -> rwConf.put(k, decryptProperty(k, v))); conf.forEach((k, v) -> rwConf.put(k, decryptProperty(k, v)));