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

init

Browse Source
This commit is contained in:
eson 2022-06-17 18:48:10 +08:00
parent 2de2796819
commit 1f482594cb
6 changed files with 48 additions and 394 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

40
pom.xml
View File

@ -3,7 +3,7 @@
<groupId>com.yuandian.dataflow</groupId> <groupId>com.yuandian.dataflow</groupId>
<artifactId>dataflow</artifactId> <artifactId>dataflow</artifactId>
<version>1.0-SNAPSHOT</version> <version>1.0.0-SNAPSHOT</version>
<packaging>jar</packaging> <packaging>jar</packaging>
<name>dataflow</name> <name>dataflow</name>
@ -22,10 +22,12 @@
<grpc.version>1.46.0</grpc.version> <grpc.version>1.46.0</grpc.version>
<slf4j.version>1.7.36</slf4j.version> <slf4j.version>1.7.36</slf4j.version>
<jraft.version>1.3.10</jraft.version> <jraft.version>1.3.10</jraft.version>
<spring.boot.web>2.7.0</spring.boot.web> <spring.boot.version>2.7.0</spring.boot.version>
<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> <snakeyaml.version>1.30</snakeyaml.version>
<yuandian.common.config.version>1.0.0-SNAPSHOT</yuandian.common.config.version>
</properties> </properties>
<dependencies> <dependencies>
@ -44,6 +46,12 @@
<scope>test</scope> <scope>test</scope>
</dependency> </dependency>
<dependency>
<groupId>com.yuandian.common</groupId>
<artifactId>config</artifactId>
<version>${yuandian.common.config.version}</version>
</dependency>
<dependency> <dependency>
<groupId>org.slf4j</groupId> <groupId>org.slf4j</groupId>
@ -77,7 +85,7 @@
<dependency> <dependency>
<groupId>org.springframework.boot</groupId> <groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId> <artifactId>spring-boot-starter-web</artifactId>
<version>${spring.boot.web}</version> <version>${spring.boot.version}</version>
</dependency> </dependency>
@ -133,14 +141,28 @@
<version>1.18.24</version> <version>1.18.24</version>
<scope>provided</scope> <scope>provided</scope>
</dependency> </dependency>
</dependencies> </dependencies>
<build> <repositories>
<repository>
<id>yuandian-nexus</id>
<name>Team Nexus Repository</name>
<url>http://mvn.yuandian.com/repository/maven-public</url>
<snapshots>
<enabled>true</enabled>
<updatePolicy>always</updatePolicy>
</snapshots>
</repository>
</repositories>
<pluginRepositories>
<pluginRepository>
<id>yuandian-nexus</id>
<name>Team Nexus Repository</name>
<url>http://mvn.yuandian.com/repository/maven-public</url>
</pluginRepository>
</pluginRepositories>
<build>
<extensions> <extensions>
<extension> <extension>
<groupId>kr.motd.maven</groupId> <groupId>kr.motd.maven</groupId>
@ -183,7 +205,7 @@
<plugin> <plugin>
<groupId>org.springframework.boot</groupId> <groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId> <artifactId>spring-boot-maven-plugin</artifactId>
<version>${spring.boot.web}</version> <version>${spring.boot.version}</version>
<executions> <executions>
<execution> <execution>
<goals> <goals>

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

@ -1,6 +1,7 @@
package com.yuandian.dataflow; package com.yuandian.dataflow;
import com.yuandian.common.Config;
import com.yuandian.dataflow.statemachine.RaftClosure; import com.yuandian.dataflow.statemachine.RaftClosure;
import com.yuandian.dataflow.statemachine.StateMachine; import com.yuandian.dataflow.statemachine.StateMachine;
@ -69,6 +70,9 @@ 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"};
String[] sprPeers = new String[]{"3440","3441","3442"}; String[] sprPeers = new String[]{"3440","3441","3442"};

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

@ -1,270 +0,0 @@
/**
* description
*
* @author eson
*2022年6月13日-17:08:46
*/
package com.yuandian.dataflow.config;
import java.io.IOException;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.Executor;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Function;
import org.yaml.snakeyaml.Yaml;
import com.alibaba.nacos.api.NacosFactory;
import com.alibaba.nacos.api.PropertyKeyConst;
import com.alibaba.nacos.api.config.ConfigService;
import com.alibaba.nacos.api.config.listener.Listener;
import com.alibaba.nacos.api.exception.NacosException;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
/**
* description nacos配置.
*
* @author eson
*2022年6月13日-17:08:46
*/
@Slf4j
@Getter
@Setter
public class Config {
public static String DEFAULT_CONFIT_FILE = "application.properties";
public static String DEFAULT_CONFIG_ADDR = "yuandian.dataflow.config.nacos.server.addr";
// public static String DEFAULT_CONFIG_DATAID = "yuandian.dataflow.config.nacos.dataid";
// public static String DEFAULT_CONFIG_GROUP = "yuandian.dataflow.config.nacos.group";+
// 默认
public static String DEFAULT_GROUP_DATAID = "yuandian.dataflow";
// 所有生成的nacos客户端
private static HashMap<String,Config> configDict = new HashMap<>();
// 配置的所有值主类
public Map<String,Object> data;
// nacos地址
public String serverAddr ;
// nacos dataId
public String dataId ;
// nacos group
public String group ;
// 线程安全配置锁
private Lock datalock;
// nacos 客户端类
private ConfigService configService;
private Config(String GroupAndDataId) throws Exception {
String[] gad = GroupAndDataId.split("\\.");
if(gad.length != 2) {
throw new Exception("Group 或者 DataId 不能存在 '.' 的命令");
}
this.group = gad[0] + ENV_TEST;
this.dataId = gad[1] + ENV_TEST;
connect();
}
/**
* 连接nacos
* @throws IOException
* @throws NacosException
*/
private void connect() throws IOException, NacosException {
if(configService != null) {
configService.shutDown();
configService = null;
}
// 获取 app
Properties prop = new Properties();
prop.load(Config.class.getClassLoader().getResourceAsStream(DEFAULT_CONFIT_FILE));
serverAddr = prop.getProperty(DEFAULT_CONFIG_ADDR + ENV_TEST).trim();
// dataId = prop.getProperty(DEFAULT_CONFIG_DATAID).trim();
// group = prop.getProperty(DEFAULT_CONFIG_GROUP).trim();
Properties properties = new Properties();
properties.put(PropertyKeyConst.SERVER_ADDR, serverAddr);
configService = NacosFactory.createConfigService(properties);
String content = configService.getConfig(dataId, group, 5000);
Yaml yaml = new Yaml();
data = yaml.load(content);
log.info(content);
datalock = new ReentrantLock();
// 监听 配置更新事件
configService.addListener(dataId, group, new Listener() {
@Override
public void receiveConfigInfo(String configInfo) {
log.debug("recieve:" + configInfo);
try {
datalock.lock();
data = (Map<String, Object>)new Yaml().load(configInfo);
log.debug("{}",data);
} finally {
datalock.unlock();
}
}
@Override
public Executor getExecutor() {
return null;
}
});
}
/**
* 根据多个key获取yaml的值 keys 路径
* @param keys 获取的key值
* @return
*/
public Object get(String ...keys) {
var cur = data;
for(var i = 0; i < keys.length - 1;i++ ) {
var key = keys[i];
cur = (Map<String, Object>) cur.get(key);
}
return cur.get(keys[keys.length - 1]);
}
/**
*
* 用于定位keys的路径后的操作. 创建keys后赋值. 如果存在keys, 可以直接赋值, 不存在则直接报错
*
* @author eson
*2022年6月15日-下午12:06:37
*/
public class Operator {
Config config;
String[] keys;
Operator(Config config, String[] keys) {
this.config = config;
this.keys = keys;
}
/**
* 创建seek的key
* @return
*/
public Operator createKeys() {
var cur = config.data;
for(var i = 0; i < keys.length;i++ ) {
var key = keys[i];
var vobj = cur.get(key);
if (vobj == null) {
vobj = new LinkedHashMap<>();
cur.put(key, vobj);
}
cur = (Map<String, Object>) vobj;
}
return this;
}
/**
* 定位后赋值
* @param value
*/
public void set(Object value) {
var cur = config.data;
for(var i = 0; i < keys.length - 1;i++ ) {
var key = keys[i];
cur = (Map<String, Object>) cur.get(key);
}
cur.put(keys[keys.length - 1], value);
}
}
/**
* 定位. eg. seek("key1", "key2")
* @param keys
* @return
*/
public Operator seek(String ...keys) {
return new Operator(this, keys);
}
/**
* 删除 key的值. 类型map操作. keys是一个路径 remove("a","b") --> {"a": {"b": 1}} 删除b
* @param keys
* @return
*/
public Object remove(String ...keys) {
var cur = data;
for(var i = 0; i < keys.length - 1;i++ ) {
var key = keys[i];
cur = (Map<String, Object>) cur.get(key);
}
return cur.remove(keys[keys.length - 1]);
}
/**
* 更新配置
* @return 返回是否发布成功.
* @throws NacosException
*/
public Boolean update() throws NacosException {
return configService.publishConfig(dataId, group,new Yaml().dumpAsMap(data));
}
public static String ENV_TEST = "";
/**
* 统一使用配置的入口函数 线程安全
* @param GroupAndDataID 使用配置的标签 eg."group.dataId"
* @param execute 匿名函数
* @return 函数返回的值, 如果不需要直接返回null
* @throws Exception
*/
public static Object UseConfig(String GroupAndDataID, Function<Config, Object> execute) throws Exception {
Config cnf;
synchronized(configDict) {
cnf = configDict.get(GroupAndDataID);
if(cnf == null) {
cnf = new Config(GroupAndDataID);
configDict.put(GroupAndDataID, cnf);
}
}
try {
cnf.datalock.lock();
var res = execute.apply(cnf);
return res;
} catch (Exception e) {
throw e;
} finally {
cnf.datalock.unlock();
}
}
/**
* 统一使用配置的入口函数 线程安全
* @param execute 统一使用配置的入口匿名方法
* @return 函数返回的值, 如果不需要直接返回null
* @throws Exception
*/
public static Object UseConfig(Function<Config, Object> execute) throws Exception {
return UseConfig(DEFAULT_GROUP_DATAID, execute);
}
}

11
src/main/java/com/yuandian/dataflow/grpc/CollectPackets.java
View File

@ -11,6 +11,7 @@ import java.time.Instant;
import com.google.protobuf.Any; import com.google.protobuf.Any;
import com.google.protobuf.InvalidProtocolBufferException; import com.google.protobuf.InvalidProtocolBufferException;
import com.yuandian.common.Config;
import com.yuandian.dataflow.proto.CollectPacketsServerGrpc; import com.yuandian.dataflow.proto.CollectPacketsServerGrpc;
@ -37,7 +38,15 @@ import lombok.extern.slf4j.Slf4j;
public class CollectPackets extends CollectPacketsServerImplBase { public class CollectPackets extends CollectPacketsServerImplBase {
public static void main(String[] args) throws InvalidProtocolBufferException { public static void main(String[] args) throws Exception {
Config.UseConfig( (cnf) -> {
var tid = cnf.get("test_id");
log.info("{}",tid);
return null;
});
// private final ManagedChannelBuilder<?> managedChannelBuilder; // private final ManagedChannelBuilder<?> managedChannelBuilder;
// private final CollectPacketsServerStub blockingStub; // private final CollectPacketsServerStub blockingStub;

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

@ -1,6 +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.server.addr=192.168.1.113:8848 # 不能 "为结尾"
yuandian.dataflow.config.nacos.group=yuandian

111
src/test/java/com/yuandian/dataflow/config/ConfigTest.java
View File

@ -1,111 +0,0 @@
package com.yuandian.dataflow.config;
import java.io.IOException;
import java.time.Instant;
import org.junit.Assert;
import org.junit.FixMethodOrder;
import org.junit.experimental.theories.suppliers.TestedOn;
import org.junit.jupiter.api.Order;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestMethodOrder;
import com.alibaba.nacos.api.exception.NacosException;
import lombok.extern.slf4j.Slf4j;
@Slf4j
public class ConfigTest {
/**
* 测试配置基础用法
* @throws Exception
*/
@Test
@Order(1)
void testUseConfig() throws Exception {
Config.ENV_TEST = "-test";
Config.UseConfig((cnf) -> {
log.info("{}",cnf.data);
Assert.assertEquals(cnf.get("key1", "key2"), "key_path");
Instant now = Instant.now();
cnf.data.put("use_config", now.toString());
try {
log.info("{}",cnf.update());
} catch (NacosException e) {
e.printStackTrace();
}
return null;
});
}
@Test
@Order(2)
void testRemove() throws Exception {
Config.ENV_TEST = "-test";
Config.UseConfig((cnf) -> {
cnf.remove("create1", "create2");
try {
log.info("{}",cnf.update()); // 所有增加删除操作要最后同步到nacos. 都需要update
} catch (NacosException e) {
e.printStackTrace();
}
cnf.remove("create1");
try {
log.info("{}",cnf.update()); // 所有增加删除操作要最后同步到nacos. 都需要update
} catch (NacosException e) {
e.printStackTrace();
}
return null;
});
}
@Test
@Order(1)
void testCreateKeys() throws Exception {
Config.ENV_TEST = "-test";
Config.UseConfig((cnf) -> {
log.info("{}",cnf.data);
cnf.seek("create1", "create2", "create3").createKeys().set(Instant.now().toString());;
try {
log.info("{}",cnf.update()); // 所有增加删除操作要最后同步到nacos. 都需要update
} catch (NacosException e) {
e.printStackTrace();
}
return null;
});
}
@Test
void testUpdate() throws Exception {
Config.ENV_TEST = "-test";
Config.UseConfig((cnf) -> {
Instant now = Instant.now();
cnf.data.put("use_config", now.toString());
try {
log.info("{}",cnf.update());
} catch (NacosException e) {
e.printStackTrace();
}
return null;
});
}
@Test
void testLabelConfig() throws Exception {
Config.UseConfig("org.fortest", (cnf)->{
log.info("{}", cnf.get("test"));
Assert.assertEquals(cnf.get("test"), "groupAndDataId");
return null;
});
}
}