test/dataflow
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

TODO: 解决配置 一体问题

Browse Source
This commit is contained in:
eson 2022-06-14 18:17:01 +08:00
parent 3d7ffe0c52
commit b7a5ea5570
4 changed files with 175 additions and 60 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
pom.xml
View File

@ -25,6 +25,7 @@
<spring.boot.web>2.7.0</spring.boot.web> <spring.boot.web>2.7.0</spring.boot.web>
<mongo.driver.version>3.12.11</mongo.driver.version> <mongo.driver.version>3.12.11</mongo.driver.version>
<nacos.version>2.1.0</nacos.version> <nacos.version>2.1.0</nacos.version>
<snakeyaml.version>1.30</snakeyaml.version>
</properties> </properties>
<dependencies> <dependencies>
@ -50,6 +51,14 @@
<version>${slf4j.version}</version> <version>${slf4j.version}</version>
</dependency> </dependency>
<dependency>
<groupId>org.yaml</groupId>
<artifactId>snakeyaml</artifactId>
<version>${snakeyaml.version}</version>
</dependency>
<dependency> <dependency>
<groupId>com.alipay.sofa</groupId> <groupId>com.alipay.sofa</groupId>
<artifactId>jraft-core</artifactId> <artifactId>jraft-core</artifactId>

2
src/main/java/com/yuandian/dataflow/Server.java
View File

@ -50,6 +50,7 @@ import com.alipay.sofa.jraft.option.ReplicatorGroupOptions;
* *
*/ */
@SpringBootApplication @SpringBootApplication
@SpringBootConfiguration
public class Server { public class Server {
@Autowired @Autowired
@ -65,6 +66,7 @@ public class Server {
} }
public static void main(String[] args) { public static void main(String[] args) {
String[] peers = new String[]{"localhost:4440","localhost:4441","localhost:4442"}; String[] peers = new String[]{"localhost:4440","localhost:4441","localhost:4442"};

218
src/main/java/com/yuandian/dataflow/config/Config.java
View File

@ -9,11 +9,25 @@ package com.yuandian.dataflow.config;
import java.io.FileInputStream;
import java.io.IOException; import java.io.IOException;
import java.io.PrintWriter;
import java.io.StringBufferInputStream; import java.io.StringBufferInputStream;
import java.io.StringReader; import java.io.StringReader;
import java.io.StringWriter;
import java.util.Map;
import java.util.Properties; import java.util.Properties;
import java.util.concurrent.Executor; import java.util.concurrent.Executor;
import java.util.function.Function;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.boot.system.SystemProperties;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.PropertySource;
import org.springframework.stereotype.Component;
import org.yaml.snakeyaml.Yaml;
import com.alibaba.nacos.api.NacosFactory; import com.alibaba.nacos.api.NacosFactory;
import com.alibaba.nacos.api.PropertyKeyConst; import com.alibaba.nacos.api.PropertyKeyConst;
import com.alibaba.nacos.api.config.ConfigService; import com.alibaba.nacos.api.config.ConfigService;
@ -21,41 +35,62 @@ import com.alibaba.nacos.api.config.listener.Listener;
import com.alibaba.nacos.api.exception.NacosException; import com.alibaba.nacos.api.exception.NacosException;
import com.google.api.client.http.InputStreamContent; import com.google.api.client.http.InputStreamContent;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
/** /**
* description * description
* *
* @author eson * @author eson
*2022年6月13日-17:08:46 *2022年6月13日-17:08:46
*/ */
@Slf4j
@Getter
@Setter
public class Config { public class Config {
private Config() throws NacosException, IOException {
String serverAddr = "192.168.1.113:8848";
String dataId = "dataflow";
String group = "yuandian";
public Map<String,Object> configMap;
@Value("${yuandian.dataflow.config.nacos.server.addr}")
public String serverAddr ;
public String dataId ;
public String group ;
private static Config config;
private ConfigService configService;
private Config() throws NacosException, IOException {
Properties prop = new Properties();
serverAddr = prop.getProperty("yuandian.dataflow.config.nacos.server.addr");
dataId = prop.getProperty("yuandian.dataflow.config.nacos.dataid") ;
group = prop.getProperty("yuandian.dataflow.config.nacos.group");
Properties properties = new Properties(); Properties properties = new Properties();
properties.put(PropertyKeyConst.SERVER_ADDR, serverAddr); properties.put(PropertyKeyConst.SERVER_ADDR, serverAddr);
ConfigService configService = NacosFactory.createConfigService(properties); configService = NacosFactory.createConfigService(properties);
String content = configService.getConfig(dataId, group, 5000); String content = configService.getConfig(dataId, group, 5000);
Yaml yaml = new Yaml();
configMap = yaml.load(content);
log.info(content);
Prop.load(new StringReader(content));
System.out.println(cnf);
configService.addListener(dataId, group, new Listener() { configService.addListener(dataId, group, new Listener() {
@Override @Override
public void receiveConfigInfo(String configInfo) { public void receiveConfigInfo(String configInfo) {
System.out.println("recieve:" + configInfo); System.out.println("recieve:" + configInfo);
try { synchronized(configMap){
Prop.load(new StringReader(configInfo)); configMap = (Map<String, Object>)new Yaml().load(configInfo);
System.out.println(cnf); System.out.println(configMap);
} catch (IOException e) {
e.printStackTrace();
} }
} }
@ -64,60 +99,127 @@ public class Config {
return null; return null;
} }
}); });
// boolean isPublishOk = configService.publishConfig(dataId, group, Prop.toString());
// System.out.println(isPublishOk);
// Thread.sleep(3000);
// content = configService.getConfig(dataId, group, 5000);
// System.out.println(content);
// boolean isRemoveOk = configService.removeConfig(dataId, group);
// System.out.println(isRemoveOk);
// Thread.sleep(3000);
// content = configService.getConfig(dataId, group, 5000);
// System.out.println(content);
// Thread.sleep(300000);
} }
public Properties Prop ; /**
* 更新配置
* @return 返回是否发布成功.
* @throws NacosException
*/
public Boolean update() throws NacosException {
return configService.publishConfig(dataId, group,new Yaml().dumpAsMap(configMap));
}
private static Config cnf ;
public Config(Properties prop) {
Prop = prop;
}
public static Config GetConfig() throws NacosException, IOException {
if(cnf == null) {
cnf = new Config();
}
return cnf;
}
/**
*
* @param execute 统一使用配置的入口
* @return 函数返回的值, 如果不需要直接返回null
* @throws NacosException
* @throws IOException
*/
public static Object UseConfig(Function<Config, Object> execute) throws NacosException, IOException {
if(config == null) {
config = new Config();
}
synchronized(config) {
return execute.apply(config);
}
}
// public Properties Prop ;
// private static Config cnf ;
// public Config(Properties prop) {
// Prop = prop;
// }
// public static Config GetConfig() throws NacosException, IOException {
// if(cnf == null) {
// cnf = new Config();
// }
// return cnf;
// }
public static void main(String[] args) throws NacosException, InterruptedException, IOException { public static void main(String[] args) throws NacosException, InterruptedException, IOException {
String serverAddr = "192.168.1.113:8848"; // String serverAddr = "192.168.1.113:8848";
String dataId = "dataflow"; // String dataId = "dataflow";
String group = "yuandian"; // String group = "yuandian";
// Properties properties = new Properties();
// properties.put(PropertyKeyConst.SERVER_ADDR, serverAddr);
// ConfigService configService = NacosFactory.createConfigService(properties);
// String content = configService.getConfig(dataId, group, 5000);
// Yaml yaml = new Yaml();
// cnf = yaml.load(content);
// log.info(content);
// // Properties cnf = new Properties();
// // cnf.load(new StringReader(content));
// System.out.println(cnf);
// configService.addListener(dataId, group, new Listener() {
// @Override
// public void receiveConfigInfo(String configInfo) {
// System.out.println("recieve:" + configInfo);
// synchronized(cnf){
// cnf = (Map<String, Object>)new Yaml().load(configInfo);
// }
// System.out.println(cnf);
// }
// @Override
// public Executor getExecutor() {
// return null;
// }
// });
Properties properties = new Properties();
properties.put(PropertyKeyConst.SERVER_ADDR, serverAddr);
ConfigService configService = NacosFactory.createConfigService(properties);
String content = configService.getConfig(dataId, group, 5000);
Properties cnf = new Properties();
cnf.load(new StringReader(content)); // cnf.put("publish", "123423112332");
System.out.println(cnf); // log.info(yaml.dumpAsMap(cnf));
configService.addListener(dataId, group, new Listener() {
@Override
public void receiveConfigInfo(String configInfo) {
System.out.println("recieve:" + configInfo);
try {
cnf.load(new StringReader(configInfo));
System.out.println(cnf);
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
@Override Config.UseConfig((cnf) -> {
public Executor getExecutor() { log.info("{}",cnf.configMap);
return null; cnf.configMap.put("use_config", "12");
try {
log.info("{}",cnf.update());
} catch (NacosException e) {
e.printStackTrace();
} }
return null;
}); });
// boolean isPublishOk = configService.publishConfig(dataId, group, yaml.dumpAsMap(cnf));
// System.out.println(isPublishOk);
Thread.sleep(300000); Thread.sleep(300000);

6
src/main/resources/application.properties
View File

@ -1,4 +1,6 @@
server.port=3440 server.port=3440
yuandian.dataflow.config.nacos.server.addr="192.168.1.113:8848"
yuandian.dataflow.config.nacos.dataid=dataflow
yuandian.dataflow.config.nacos.group=yuandian