Skip to content

D
Distributed Computing and Network Fusion System
Commit 0e7f587b authored by 谢建斌's avatar 谢建斌
Browse files
Options

代码重构

parent 29112de4
Expand all Hide whitespace changes
Inline Side-by-side
Showing with 459 additions and 629 deletions
apps/cnf-spaceiot-longConnection/csl-node-ground/src/main/java/top/ninwoo/ground/register/MyApplicationRunner.java
View file @ 0e7f587b
......@@ -10,11 +10,9 @@ import top.ninwoo.common.entity.DataPackage;
import top.ninwoo.ground.service.GroundService;
import top.ninwoo.ground.service.IPService;
import javax.xml.crypto.Data;
import java.io.*;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.List;
@Component
public class MyApplicationRunner implements ApplicationRunner {
......@@ -22,11 +20,11 @@ public class MyApplicationRunner implements ApplicationRunner {
private static final String fileName = "route_path.txt"; //生成的路由保存文件
private static final String timeSlot = "35"; //时隙,可修改
private static final String timeSlot = "0"; //时隙,可修改
private static final String sourceNode = "ground3";
private static final String sourceNode = "ground0";
private static final String destinationNode = "groung14";
private static final String destinationNode = "ground7";
@Autowired
GroundService groundService;
......@@ -34,100 +32,104 @@ public class MyApplicationRunner implements ApplicationRunner {
@Autowired
IPService ipService;
@Autowired
DataPackage dataPackage;
/*================================订阅节点的run函数=============================================*/
@Override
public void run(ApplicationArguments args) throws Exception {
logger.info("BeiJing Node Initiate a request");
dataPackage.setKey("temperature");
dataPackage.setSubNode(sourceNode);
dataPackage.setPubNode(destinationNode);
groundService.execPy(timeSlot, sourceNode, destinationNode);
String nextHop = groundService.readFileContent(fileName);
String nextIp = ipService.getIpByAppName(nextHop);
send(dataPackage, nextIp, 2021);
// 接收回传信息
ServerSocket serverSocket = new ServerSocket(8900);
//获取数据流
Socket accept = serverSocket.accept();
InputStream inputStream = accept.getInputStream();
//读取数据流
ObjectInputStream objectInputStream = new ObjectInputStream(new BufferedInputStream(inputStream));
Object object = null;
try {
object = objectInputStream.readObject();
DataPackage receiveData = (DataPackage)object;
//输出数据流
logger.info("BeiJing Node Receive Data");
logger.info(receiveData.getKey() + ":" + receiveData.getValue());
} catch (ClassNotFoundException e) {
e.printStackTrace();
}
serverSocket.close();
accept.close();
System.out.println();
}
/*================================其他节点的run函数=============================================*/
/*================================源节点的run函数=============================================*/
// @Override
// public void run(ApplicationArguments args) throws Exception {
// ServerSocket serverSocket = null;
// InputStream inputStream = null;
// ObjectInputStream objectInputStream = null;
// try {
// serverSocket = new ServerSocket(2021);
// while (true) {
// Socket client = serverSocket.accept();
// inputStream = client.getInputStream();
// objectInputStream = new ObjectInputStream(new BufferedInputStream(inputStream));
// Object object = objectInputStream.readObject();
// DataPackage dataPackage = (DataPackage) object;
// logger.info("NanSha Node Initiate a request");
// DataPackage dataPackage = new DataPackage();
// dataPackage.setKey("temperature");
// dataPackage.setSubNode(sourceNode);
// dataPackage.setPubNode(destinationNode);
//
// // 获取当前节点剩余跳数，若为0，则当前节点为目的节点；若不为0，则当前节点为进行转发的地面节点
// int remainingHops = dataPackage.getRemainingHops() - 1;
// if (remainingHops == 0) {
// logger.info("到达消息拥有节点，获取消息然后发送至消息订阅节点");
// dataPackage.setValue(String.valueOf(groundService.getRandom()));
// groundService.execPy(timeSlot, destinationNode, sourceNode);
// String nextHop = groundService.readFileContent(fileName);
// String nextIp = ipService.getIpByAppName(nextHop);
// groundService.execPy(timeSlot, sourceNode, destinationNode);
// String[][] newRoute = groundService.readFileContent(fileName);
// String nextHop = newRoute[0][1];
// String nextIp = ipService.getIpByAppName(nextHop);
//
// send(dataPackage, nextIp, 2021);
// continue;
// } else {
// logger.info("中间地面节点，转发至下一节点");
// dataPackage.setRemainingHops(remainingHops);
// String[][] route = dataPackage.getRoute();
// String nextHop = route[0][dataPackage.getTotalHops() - remainingHops + 1];
// String nextIp = ipService.getIpByAppName(nextHop);
// send(dataPackage, nextIp, 2021);
// continue;
// }
// }
// } catch (IOException | ClassNotFoundException e) {
// send(dataPackage, nextIp, 2021);
//
// // 接收回传信息
// ServerSocket serverSocket = new ServerSocket(8900);
// //获取数据流
// Socket accept = serverSocket.accept();
// InputStream inputStream = accept.getInputStream();
// //读取数据流
// ObjectInputStream objectInputStream = new ObjectInputStream(new BufferedInputStream(inputStream));
// Object object = null;
// try {
// object = objectInputStream.readObject();
// DataPackage receiveData = (DataPackage)object;
// //输出数据流
// logger.info("NanSha Node Receive Data");
// logger.info(receiveData.getKey() + ":" + receiveData.getValue());
// } catch (ClassNotFoundException e) {
// e.printStackTrace();
// } finally {
// try {
// if (objectInputStream != null) {
// objectInputStream.close();
// }
// if (inputStream != null) {
// inputStream.close();
// }
// if (serverSocket != null) {
// serverSocket.close();
// }
// } catch (IOException e) {
// e.printStackTrace();
// }
// }
// serverSocket.close();
// accept.close();
// System.out.println();
// }
/*================================目的节点的run函数=============================================*/
@Override
public void run(ApplicationArguments args) throws Exception {
logger.info("节点已启动");
ServerSocket serverSocket = null;
InputStream inputStream = null;
ObjectInputStream objectInputStream = null;
try {
serverSocket = new ServerSocket(2021);
while (true) {
Socket client = serverSocket.accept();
inputStream = client.getInputStream();
objectInputStream = new ObjectInputStream(new BufferedInputStream(inputStream));
Object object = objectInputStream.readObject();
DataPackage dataPackage = (DataPackage) object;
// 获取当前节点剩余跳数，若为0，则当前节点为目的节点；若不为0，则当前节点为进行转发的地面节点（当前没有中间地面节点）
int remainingHops = dataPackage.getRemainingHops() - 1;
if (remainingHops == 0) {
logger.info("到达目的节点，获取消息然后发送至消息订阅节点");
dataPackage.setValue(String.valueOf(groundService.getRandom()));
groundService.execPy(timeSlot, destinationNode, sourceNode);
String[][] newRoute = groundService.readFileContent(fileName);
dataPackage.setRoute(newRoute);
dataPackage.setRemainingHops(Integer.valueOf(newRoute[3][0]));
dataPackage.setTotalHops(Integer.valueOf(newRoute[3][0]));
String nextHop = newRoute[0][1];
String nextIp = ipService.getIpByAppName(nextHop);
send(dataPackage, nextIp, 2021);
continue;
} else {
logger.info("中间地面节点，转发至下一节点");
dataPackage.setRemainingHops(remainingHops);
String[][] route = dataPackage.getRoute();
String nextHop = route[0][dataPackage.getTotalHops() - remainingHops + 1];
String nextIp = ipService.getIpByAppName(nextHop);
send(dataPackage, nextIp, 2021);
continue;
}
}
} catch (IOException | ClassNotFoundException e) {
e.printStackTrace();
} finally {
try {
if (objectInputStream != null) {
objectInputStream.close();
}
if (inputStream != null) {
inputStream.close();
}
if (serverSocket != null) {
serverSocket.close();
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
/**
* 发送数据包到下一节点
......
apps/cnf-spaceiot-longConnection/csl-node-ground/src/main/java/top/ninwoo/ground/service/GroundService.java
View file @ 0e7f587b
......@@ -2,31 +2,26 @@ package top.ninwoo.ground.service;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import top.ninwoo.common.entity.DataPackage;
import org.springframework.stereotype.Service;
import java.io.*;
import java.util.ArrayList;
import java.util.Random;
@Service
public class GroundService {
private static final Logger logger = LoggerFactory.getLogger(GroundService.class);
private static final String pyFilePath = "E:\\xjb\\xjb.py"; //如果未指定.py文件的完全路径，则默认从工程当前目录下搜索
private static final String pyAllPath = "E:\\xjb"; //注意：当命令行参数分开写的时候，exec后面不用添加一个空格。当命令行参数一起写的时候，exe后面一定要添加一个空格
private static final String pyRoutePath = "E:\\xjb";
private static final String pyFilePath = "/luyou/xjb.py"; //如果未指定.py文件的完全路径，则默认从工程当前目录下搜索
private static final String pyAllPath = "/luyou/"; //注意：当命令行参数分开写的时候，exec后面不用添加一个空格。当命令行参数一起写的时候，exe后面一定要添加一个空格
private static final String pyRoutePath = "/luyou/";
private static Process proc = null;//java进程类
@Autowired
DataPackage dataPackage;
@Autowired
GroundService groundService;
/**
* 执行*.py文件
*/
public void execPy(String timeSlot, String sourceNode, String destinationNode) {
String[] arguments = new String[] {"py", pyFilePath, pyAllPath, pyRoutePath, timeSlot, sourceNode, destinationNode};
String[] arguments = new String[] {"python3", pyFilePath, pyAllPath, pyRoutePath, timeSlot, sourceNode, destinationNode};
try {
proc = Runtime.getRuntime().exec(arguments);
BufferedReader in = new BufferedReader(new InputStreamReader(proc.getInputStream()));
......@@ -36,7 +31,9 @@ public class GroundService {
}
in.close();
int re = proc.waitFor();//返回0：成功。其余返回值均表示失败，如：返回错误代码1：操作不允许，表示调用python脚本失败
System.out.println(re);
if (re != 0) {
logger.error("调用python脚本失败");
}
} catch (IOException e) {
e.printStackTrace();
} catch (InterruptedException e) {
......@@ -48,12 +45,11 @@ public class GroundService {
* 读取route_path.txt生成路由表,返回下一跳节点
* @param fileName
*/
public String readFileContent(String fileName) throws IOException {
public String[][] readFileContent(String fileName) throws IOException {
/**
* 文件读入
*/
String filePath = pyRoutePath + "\\" + fileName;
// File file = new file(filePath);
String filePath = pyRoutePath + "/" + fileName;
FileReader in = null;
try {
in = new FileReader(filePath);
......@@ -75,8 +71,9 @@ public class GroundService {
e.printStackTrace();
}
list.add(s);
if(s == null)
if(s == null) {
break;
}
}
/**
......@@ -94,10 +91,10 @@ public class GroundService {
}
}
br.close();
dataPackage.setRoute(square);
dataPackage.setRemainingHops(Integer.valueOf(square[3][0]));
dataPackage.setTotalHops(Integer.valueOf(square[3][0]));
return square[0][1];
// Print(rowLength, colLength, square);
return square;
//return square[0][1];
}
/**
......@@ -114,7 +111,7 @@ public class GroundService {
public static void Print(int colLength, int rowLength, String[][] square) {
public void Print(int rowLength, int colLength, String[][] square) {
for (int i = 0; i < rowLength; i++) {
if (i > 0)
System.out.print("\n");
......@@ -122,13 +119,6 @@ public class GroundService {
System.out.print(square[i][j] + "\t");
}
}
System.out.println(Integer.valueOf(square[3][0]));
}
public static void main(String[] args) throws IOException {
// execPy();
// readFileContent("route_path.txt");
}
}
apps/cnf-spaceiot-longConnection/csl-node-ground/src/main/resources/ application.properties 0 → 100644
View file @ 0e7f587b
#项目端口
server.port=8999
#接入云端的端口设置
bishe.app.app-name=joliu
bishe.app.cloud-url=192.168.31.37:9091
apps/cnf-spaceiot-longConnection/csl-node-satellite/src/main/java/top/ninwoo/satellite/register/MyApplicationRunner.java
View file @ 0e7f587b
......@@ -21,8 +21,6 @@ public class MyApplicationRunner implements ApplicationRunner {
@Autowired
IPService ipService;
@Autowired
DataPackage dataPackage;
/**
* Callback used to run the bean.
......@@ -32,6 +30,7 @@ public class MyApplicationRunner implements ApplicationRunner {
*/
@Override
public void run(ApplicationArguments args) throws Exception {
logger.info("节点开始运行");
ServerSocket serverSocket = null;
InputStream inputStream = null;
ObjectInputStream objectInputStream = null;
......
apps/cnf-spaceiot-longConnection/csl-node-satellite/src/main/resources/application.properties 0 → 100644
View file @ 0e7f587b
#项目端口
server.port=8999
#接入云端的端口设置
bishe.app.app-name=joliu
bishe.app.cloud-url=192.168.31.37:9091
\ No newline at end of file
apps/cnf-spaceiot-mqtt/src/main/java/top/ninwoo/mqtt/mqttUtils/MyMqttClient.java
View file @ 0e7f587b
......@@ -138,7 +138,7 @@ public class MyMqttClient {
/**
* 订阅主题
*/
public static String subTopic(String topic, int qos) {
public static void subTopic(String topic, int qos) {
if (mqttClient != null && mqttClient.isConnected()) {
try {
mqttConnectOptions.setWill(topic, "close".getBytes(), 2, true);
......@@ -146,30 +146,27 @@ public class MyMqttClient {
SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");//设置日期格式
logger.info("订阅时间========" + df.format(new Date()) + " " + System.currentTimeMillis());
mqttClient.subscribe(topic, qos);
//mqttClient.setCallback(mqttClientCallback);
// MqttMessage mqttMessage = new MqttMessage();
// String messages = new String(mqttMessage.getPayload());
// return messages;
//logger.info("订阅成功");
mqttClient.setCallback(mqttClientCallback);
logger.info("订阅成功");
} catch (MqttException me) {
me.printStackTrace();
}
} else {
System.out.println("mqttClient is Error");
}
return null;
}
/**
* 订阅主题
*/
public static void subTopic(String[] topic, int[] qos) {
if (mqttClient != null && mqttClient.isConnected()) {
try {
//logger.info("clientId: " + mqttClient.getClientId() + "订阅主题" + topic);
logger.info("clientId: " + mqttClient.getClientId() + "订阅主题" + topic);
System.out.println("mqttClient subscribe topic : " + topic);
mqttClient.subscribe(topic, qos);
mqttClient.setCallback(mqttClientCallback);
//logger.info("订阅成功");
logger.info("订阅成功");
} catch (MqttException me) {
me.printStackTrace();
}
......@@ -225,9 +222,7 @@ public class MyMqttClient {
mqttMessage.setQos(qos);
mqttMessage.setRetained(retained);
mqttMessage.setPayload(message.getBytes());
MqttTopic topic = mqttClient.getTopic(pubTopic);
if (topic == null) {
logger.error("topic not exist");
}
......@@ -238,9 +233,7 @@ public class MyMqttClient {
try {
token = topic.publish(mqttMessage);
token.waitForCompletion();
//logger.info("message is published completely!" + token.isComplete());
System.out.println("message is published completely! "
+ token.isComplete());
logger.info("message is published completely!" + token.isComplete());
} catch (MqttPersistenceException me) {
logger.error("publish fail", me);
me.printStackTrace();
......
apps/cnf-spaceiot-shortConnection/css-node-ground/pom.xml
View file @ 0e7f587b
......@@ -17,11 +17,13 @@
<groupId>top.ninwoo</groupId>
<artifactId>css-common-api</artifactId>
<version>1.0-SNAPSHOT</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>top.ninwoo</groupId>
<artifactId>css-node-satellite</artifactId>
<version>1.0-SNAPSHOT</version>
<scope>compile</scope>
</dependency>
</dependencies>
......@@ -30,7 +32,7 @@
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
<version>${springboot.version}</version>
<version>1.3.7.RELEASE</version>
<configuration>
<mainClass>top.ninwoo.ground.GroundMain</mainClass>
</configuration>
......
apps/cnf-spaceiot-shortConnection/css-node-ground/src/main/java/top/ninwoo/ground/register/MyApplicationRunner.java 0 → 100644
View file @ 0e7f587b
package top.ninwoo.ground.register;
import com.google.common.collect.Lists;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;
import top.ninwoo.common.entity.ClientKVReq;
import top.ninwoo.satellite.current.SleepHelper;
import top.ninwoo.satellite.entity.LogEntry;
import top.ninwoo.satellite.rpc.Request;
import top.ninwoo.satellite.rpc.Response;
import top.ninwoo.satellite.rpc.RpcClient;
import top.ninwoo.satellite.rpc.impl.DefaultRpcClient;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;
@Component
public class MyApplicationRunner implements ApplicationRunner {
private static final Logger LOGGER = LoggerFactory.getLogger(MyApplicationRunner.class);
private final static RpcClient client = new DefaultRpcClient();
static String addr;
// static List<String> list = Lists.newArrayList("10.10.1.2:8775","10.10.1.3:8775","10.10.1.4:8775", "10.10.1.5:8775", "10.10.1.6:8775", "10.10.1.7:8775", "10.10.1.8:8775",
// "10.10.1.9:8775","10.10.1.10:8775","10.10.1.11:8775", "10.10.1.12:8775", "10.10.1.13:8775", "10.10.1.14:8775", "10.10.1.15:8775",
// "10.10.1.16:8775");
static List<String> list = Lists.newArrayList("10.10.1.2:8775","10.10.1.3:8775","10.10.1.4:8775");
/**
* Callback used to run the bean.
*
* @param args incoming application arguments
* @throws Exception on error
*/
@Override
public void run(ApplicationArguments args) throws Exception {
Thread.currentThread().sleep(30000);
LOGGER.info("客户端发起请求");
AtomicLong count = new AtomicLong(3);
for (int i = 3; ; i++) {
try {
int index = (int) (count.incrementAndGet() % list.size());
addr = list.get(index);
ClientKVReq obj = ClientKVReq.newBuilder().key("hello:" + i).value("world:" + i).type(ClientKVReq.PUT).build();
Request<ClientKVReq> r = new Request<>();
r.setObj(obj);
r.setUrl(addr);
r.setCmd(Request.CLIENT_REQ);
Response<String> response;
try {
response = client.send(r);
} catch (Exception e) {
r.setUrl(list.get((int) ((count.incrementAndGet()) % list.size())));
response = client.send(r);
}
LOGGER.info("request content : {}, url : {}, put response : {}", obj.key + "=" + obj.getValue(), r.getUrl(), response.getResult());
//SleepHelper.sleep(1000);
obj = ClientKVReq.newBuilder().key("hello:" + i).type(ClientKVReq.GET).build();
addr = list.get(index);
r.setUrl(addr);
r.setObj(obj);
Response<LogEntry> response2;
try {
response2 = client.send(r);
} catch (Exception e) {
r.setUrl(list.get((int) ((count.incrementAndGet()) % list.size())));
response2 = client.send(r);
}
LOGGER.info("request content : {}, url : {}, get response : {}", obj.key + "=" + obj.getValue(), r.getUrl(), response2.getResult());
} catch (Exception e) {
e.printStackTrace();
i = i - 1;
}
SleepHelper.sleep(5000);
}
}
}
apps/cnf-spaceiot-shortConnection/css-node-ground/src/main/java/top/ninwoo/ground/register/RaftClient.java
View file @ 0e7f587b
//package top.ninwoo.ground.register;
//
package top.ninwoo.ground.register;
//import com.alipay.remoting.exception.RemotingException;
//import com.google.common.collect.Lists;
//import org.slf4j.Logger;
......
apps/cnf-spaceiot-shortConnection/css-node-satellite/Dockerfile
View file @ 0e7f587b
......@@ -6,14 +6,11 @@ RUN yum install -y http://dl.fedoraproject.org/pub/epel/6/i386/epel-release-6-8.
RUN yum install -y net-tools.x86_64
RUN echo "net.ipv4.ip_forward=1" >> /etc/sysctl.conf
RUN mkdir -p /tmp/static
COPY jdk1.8.0_111 jdk1.8.0_111
ENV JAVA_HOME=/jdk1.8.0_111
ENV PATH=$JAVA_HOME/bin:$PATH
ENV CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar
ADD space-iot-raft-1.0-SNAPSHOT.jar sate_raft.jar
ADD getIpTest.jar selfIp.jar
ADD space-iot-raft-1.0-SNAPSHOT.jar satellite.jar
ENTRYPOINT ["java","-jar","sate_raft.jar"]
ENTRYPOINT ["java","-jar","satellite.jar"]
apps/cnf-spaceiot-shortConnection/css-node-satellite/pom.xml
View file @ 0e7f587b
......@@ -18,6 +18,7 @@
<version>1.0-SNAPSHOT</version>
<scope>compile</scope>
</dependency>
<!--mqtt-->
<dependency>
<groupId>org.springframework.integration</groupId>
......@@ -46,7 +47,7 @@
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
<version>${springboot.version}</version>
<version>1.3.7.RELEASE</version>
<configuration>
<mainClass>top.ninwoo.satellite.RaftNodeBootStrap</mainClass>
</configuration>
......
apps/cnf-spaceiot-shortConnection/css-node-satellite/src/main/java/top/ninwoo/satellite/nodeCommon/NodeConfig.java
View file @ 0e7f587b
......@@ -13,7 +13,7 @@ public class NodeConfig {
// 自身节点地址
public String selfAddr;
//所有节点地址
// 所有节点地址
public List<String> peerAddrs;
}
\ No newline at end of file
apps/cnf-spaceiot-shortConnection/css-node-satellite/src/main/java/top/ninwoo/satellite/register/MyApplicationRunner.java
View file @ 0e7f587b
......@@ -48,15 +48,18 @@ public class MyApplicationRunner implements ApplicationRunner {
NodeConfig config = new NodeConfig();
Thread.currentThread().sleep(20000);
Thread.currentThread().sleep(25000);
// 自身节点地址
config.setSelfAddr(ipServiceImpl.getSelfIp() + ":8775");
config.setSelfAddr(ipServiceImpl.getSelfIp());
// 所有节点地址
// Long clusterId = 11113l;
// config.setPeerAddrs(Arrays.asList(ipServiceImpl.getIpList(clusterId, "sate")));
String[] peerAddr = {"10.10.1.2:8775","10.10.1.3:8775","10.10.1.4:8775", "10.10.1.5:8775", "10.10.1.6:8775", "10.10.1.7:8775", "10.10.1.8:8775"};//集群列表
// String[] peerAddr = {"10.10.1.2:8775","10.10.1.3:8775","10.10.1.4:8775", "10.10.1.5:8775", "10.10.1.6:8775", "10.10.1.7:8775", "10.10.1.8:8775",
// "10.10.1.9:8775","10.10.1.10:8775","10.10.1.11:8775", "10.10.1.12:8775", "10.10.1.13:8775", "10.10.1.14:8775", "10.10.1.15:8775",
// "10.10.1.16:8775"};//集群列表
String[] peerAddr = {"10.10.1.2:8775","10.10.1.3:8775","10.10.1.4:8775"};//集群列表
config.setPeerAddrs(Arrays.asList(peerAddr));
Node node = DefaultNode.getInstance();
......
apps/cnf-spaceiot-shortConnection/css-node-satellite/src/main/java/top/ninwoo/satellite/service/impl/IPServiceImpl.java
View file @ 0e7f587b
......@@ -19,7 +19,7 @@ public class IPServiceImpl implements IPService {
@Override
public String getSelfIp() throws UnknownHostException {
InetAddress addr = Inet4Address.getLocalHost();
return addr.getHostAddress();
return addr.getHostAddress() + ":8775";
}
@Override
......
apps/cnf-spaceiot-shortConnection/css-node-satellite/src/main/java/top/ninwoo/satellite/util/LifeCycle.java
View file @ 0e7f587b
......@@ -5,4 +5,5 @@ public interface LifeCycle {
void init() throws Throwable;
void destroy() throws Throwable;
}
apps/cnf-spaceiot-shortConnection/css-node-satellite/src/main/java/top/ninwoo/satellite/util/impl/DefaultNode.java
View file @ 0e7f587b
......@@ -21,7 +21,6 @@ import top.ninwoo.satellite.rpc.RpcClient;
import top.ninwoo.satellite.rpc.RpcServer;
import top.ninwoo.satellite.rpc.impl.DefaultRpcClient;
import top.ninwoo.satellite.rpc.impl.DefaultRpcServer;
import top.ninwoo.satellite.util.mqttUtils.MyMqttClient;
import top.ninwoo.satellite.util.*;
import java.util.*;
......@@ -100,8 +99,6 @@ public class DefaultNode<T> implements Node<T>, LifeCycle, ClusterMembershipChan
public RpcClient rpcClient = new DefaultRpcClient();
public MyMqttClient myMqttClient;
public StateMachine stateMachine;
/* ============================== */
......@@ -132,7 +129,6 @@ public class DefaultNode<T> implements Node<T>, LifeCycle, ClusterMembershipChan
public void setConfig(NodeConfig config) throws MqttException {
this.config = config;
stateMachine = DefaultStateMachine.getInstance();
// myMqttClient = new MyMqttClient();
logModule = DefaultLogModule.getInstance();
peerSet = PeerSet.getInstance();
......
apps/cnf-spaceiot-shortConnection/css-node-satellite/src/main/java/top/ninwoo/satellite/util/impl/DefaultStateMachine.java
View file @ 0e7f587b
......@@ -40,7 +40,6 @@ public class DefaultStateMachine implements StateMachine {
RocksDB.loadLibrary();
}
private DefaultStateMachine() {
synchronized (this) {
try {
......
apps/cnf-spaceiot-shortConnection/css-node-satellite/src/main/java/top/ninwoo/satellite/util/mqttUtils/MqttClientCallback.java deleted 100644 → 0
View file @ 29112de4
package top.ninwoo.satellite.util.mqttUtils;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import java.io.UnsupportedEncodingException;
import java.text.SimpleDateFormat;
import java.util.Date;
/**
* 发布消息的回调类
*
* 必须实现MqttCallback的接口并实现对应的相关接口方法CallBack 类将实现 MqttCallBack。
* 每个客户机标识都需要一个回调实例。在此示例中，构造函数传递客户机标识以另存为实例数据。
* 在回调中，将它用来标识已经启动了该回调的哪个实例。
* 必须在回调类中实现三个方法：
*
* public void messageArrived(MqttTopic topic, MqttMessage message)接收已经预订的发布。
*
* public void connectionLost(Throwable cause)在断开连接时调用。
*
* public void deliveryComplete(MqttDeliveryToken token))
* 接收到已经发布的 QoS 1 或 QoS 2 消息的传递令牌时调用。
* 由 MqttClient.connect 激活此回调。
*
*/
@Slf4j
@Component
public class MqttClientCallback implements MqttCallback {
private static final Logger logger = LoggerFactory.getLogger(MqttClientCallback.class);
@Override
public void connectionLost(Throwable arg0) {
logger.info("mqtt连接断开，可以做重连");
MyMqttClient.reConnect();
}
@Override
public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
logger.info("deliveryComplete---------" + iMqttDeliveryToken.isComplete());
}
@Override
public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
String messages = new String(mqttMessage.getPayload());
if(!messages.equals("close")) {
System.out.println("{");
System.out.println("接收消息主题 : " + topic);
System.out.println("接收消息Qos : " + mqttMessage.getQos());
System.out.println("接收消息内容 : " + messages);
System.out.println("}");
SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");//设置日期格式
logger.info("接收时间========" + df.format(new Date()) + " " + System.currentTimeMillis());
}
}
public void perform(String topicP, JSONObject json) throws MqttException, UnsupportedEncodingException {
//你的业务模块
}
}
apps/cnf-spaceiot-shortConnection/css-node-satellite/src/main/java/top/ninwoo/satellite/util/mqttUtils/PublishThread.java deleted 100644 → 0
View file @ 29112de4
package top.ninwoo.satellite.util.mqttUtils;
import org.springframework.stereotype.Component;
import static top.ninwoo.satellite.util.mqttUtils.MyMqttClient.poll;
@Component
public class PublishThread implements Runnable {
@Override
public void run()
{
String[] obj = null;
System.out.println("mqtt publish thread start");
while (true) {
obj = poll();
if (obj != null) {
String topic = obj[0];
String content = obj[1];
int qos = Integer.valueOf(obj[2]);
System.out.println("mqtt从队列取出topic:" + topic + ",content:" + content + ",qos" + qos);
try {
MyMqttClient.pubMessage(1, true, topic, content);
System.out.println("发送mqtt消息,topic: "+ topic +" ,content: " + content);
} catch (Exception e) {
System.out.println("发消息给设备,topic:" + topic + ",content:" + content);
e.printStackTrace();
}
} else{
try {
Thread.sleep(1000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
}
apps/cnf-spaceiot-shortConnection/pom.xml
View file @ 0e7f587b
......@@ -114,17 +114,17 @@
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
<version>${springboot.version}</version>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<!-- <version>3.7.0</version>
<configuration>
<source>1.8</source>
<target>1.8</target>
</configuration>-->
<version>1.3.7.RELEASE</version>
</plugin>
<!-- <plugin>-->
<!-- <groupId>org.apache.maven.plugins</groupId>-->
<!-- <artifactId>maven-compiler-plugin</artifactId>-->
<!-- &lt;!&ndash; <version>3.7.0</version>-->
<!-- <configuration>-->
<!-- <source>1.8</source>-->
<!-- <target>1.8</target>-->
<!-- </configuration>&ndash;&gt;-->
<!-- </plugin>-->
</plugins>
</build>
......
cnf-client-starter/src/main/java/top/ninwoo/bishe/starter/config/ClientProperties.java
View file @ 0e7f587b
......@@ -6,9 +6,8 @@ import org.springframework.boot.context.properties.ConfigurationProperties;
@Data
@ConfigurationProperties(prefix = "bishe.app")
public class ClientProperties {
private String zookeeperUrl = "zookeeper://127.0.0.1:2181";
private String cloudUrl = "127.0.0.1:9300";
private String zookeeperUrl = "zookeeper://192.168.31.37:2181";
private String cloudUrl = "192.168.31.37:9091";
private String appName = "default";
}
cnf-edge-center/src/main/java/top/ninwoo/edgecenter/service/impl/ClusterServiceImpl.java
View file @ 0e7f587b
......@@ -106,9 +106,17 @@ public class ClusterServiceImpl implements ClusterService {
// 创建docker的逻辑
// TODO: 根据Mode选择不同的模式
// 根据副本数量创建多少个容器
if(c.getReplicas() <= 0) {
// do nothing
} else {
if(c.getReplicas() == 1) {
DockerContainer containerConfig = c.getDockerContainer();
String containerName = containerConfig.getName();
// 校验Container名称是否唯一
if(clustersInfo.get(clusterConfig.getId()).containsKey(containerName)) {
throw new RuntimeException("Container Name must be unique!");
}
clustersInfo.get(clusterConfig.getId()).put(containerName, new HashSet<>());
DockerContainer dockerContainer = dockerService.runDocker(containerConfig);
clustersInfo.get(clusterConfig.getId()).get(containerName).add(dockerContainer.getId());
} else if (c.getReplicas() > 1) {
DockerContainer containerConfig = c.getDockerContainer();
String containerName = containerConfig.getName();
// 校验Container名称是否唯一
......@@ -119,7 +127,7 @@ public class ClusterServiceImpl implements ClusterService {
for (int i = 0; i < c.getReplicas(); i++) {
// 处理DockerContainer中的名称
if(!"".equals(containerConfig.getName())) {
containerConfig.setName(containerName+ "_" + i);
containerConfig.setName(containerName+ "" + i);
}
DockerContainer dockerContainer = dockerService.runDocker(containerConfig);
clustersInfo.get(clusterConfig.getId()).get(containerName).add(dockerContainer.getId());
......
cnf-test/src/test/java/top/ninwoo/JbxieLongConTest.java 0 → 100644
View file @ 0e7f587b
package top.ninwoo;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
import top.ninwoo.bishe.starter.service.ClusterService;
import top.ninwoo.bishe.starter.service.NetworkService;
import top.ninwoo.client.BisheTestMain;
import top.ninwoo.common.entity.*;
import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
@RunWith(SpringRunner.class)
@SpringBootTest(classes = BisheTestMain.class)
public class JbxieLongConTest {
@Autowired
private ClusterService clusterService;
@Resource
private NetworkService networkService;
@Test
public void testSendSeparateConfig() {
/*source节点*/
ContainerDescription containerDescriptionSource = new ContainerDescription();
containerDescriptionSource.setMode("normal");
containerDescriptionSource.setReplicas(1);
DockerContainer container1 = new DockerContainer();
container1.setName("ground0");//创建一个source容器
container1.setCommand("sh");
container1.setImage("source");
// container1.setPorts("2021");//指定容器端口
containerDescriptionSource.setDockerContainer(container1);
/*destination节点*/
ContainerDescription containerDescriptionDestination = new ContainerDescription();
containerDescriptionDestination.setMode("normal");
containerDescriptionDestination.setReplicas(1);
DockerContainer container2 = new DockerContainer();
container2.setName("ground7");//创建一个destination容器
container2.setCommand("sh");
container2.setImage("destination");
// container2.setPorts("2021");//指定容器端口
containerDescriptionDestination.setDockerContainer(container2);
/*satellite*/
ContainerDescription containerDescriptionSatellite = new ContainerDescription();
containerDescriptionSatellite.setMode("normal");
containerDescriptionSatellite.setReplicas(30);
DockerContainer container3 = new DockerContainer();
container3.setName("sate");//创建sate容器
container3.setCommand("sh");
container3.setImage("satellite");
// container3.setPorts("2021");//指定容器端口
containerDescriptionSatellite.setDockerContainer(container3);
ArrayList<SeparatedClusterConfig> clusterConfigs = new ArrayList<>();
SeparatedClusterConfig separatedClusterConfig = new SeparatedClusterConfig();
separatedClusterConfig.setEdgeNodeId("192.168.31.213:18088"); //边缘节点IP+端口号
List<ContainerDescription> cds = new ArrayList<>();
cds.add(containerDescriptionSource);//将source容器添加到list集合中
cds.add(containerDescriptionDestination);//将destination容器添加到list集合中
cds.add(containerDescriptionSatellite);//将satellite容器添加到list集合中
ClusterConfig clusterConfigCom = new ClusterConfig();
clusterConfigCom.setId(11113l);
clusterConfigCom.setOwner("jbxie");
clusterConfigCom.setDockers(cds);
NetworkTopology topo = new NetworkTopology();
topo.setAppNames(new String[]{"ground0","ground7", "sate", "br:ovs1", "br:ovs2"});
topo.setTopologyId(11);
topo.setTopology(new int[][]{{0,0,0,0,0},{0,0,0,0,0},{0,0,0,0,0},{1,1,0,0,0},{0,1,1,0,0}});
clusterConfigCom.setTopology(topo);
separatedClusterConfig.setClusterConfig(clusterConfigCom);
clusterConfigs.add(separatedClusterConfig);
clusterService.sendClusterConfigToEdgeNode(clusterConfigs);
}
//测试容器ip
@Test
public void getIpListByAppNameTest() {
List<String> ipList = networkService.getIpListByAppName(11113L, "sate");
System.out.println(ipList);
}
@Test
public void getIpList() {
List<String> ipListS = networkService.getIpListByAppName(11113L, "sate");
String[] ipList = new String[ipListS.size()];
for (int i = 0; i < ipListS.size(); i++) {
if(!ipListS.isEmpty()){
String ip_tmp = ipListS.get(i);
String[] split_list = ip_tmp.split("/");
ipList[i] = split_list[0]+":8082";
}
}
for (int j = 0; j < ipList.length; j++) {
System.out.println(ipList[j]);
}
}
@Test
public void removeClusterFromEdgeNodeTest() {
clusterService.removeClusterFromEdgeNode(11113L);
}
@Test
public void getLogicalNetworkTopologyTest() {
NetworkTopology logicalNetworkTopology = networkService.getLogicalNetworkTopology(11113L);
System.out.println(logicalNetworkTopology);
}
}
cnf-test/src/test/java/top/ninwoo/JbxieRaftTests.java
View file @ 0e7f587b
......@@ -28,39 +28,40 @@ public class JbxieRaftTests {
ArrayList<SeparatedClusterConfig> clusterConfigs = new ArrayList<>();
SeparatedClusterConfig separatedClusterConfig = new SeparatedClusterConfig();
// TODO: 这个ID应该是从借口获取的
separatedClusterConfig.setEdgeNodeId("192.168.31.171:18088"); //边缘节点IP+端口号
separatedClusterConfig.setEdgeNodeId("192.168.31.213:18088"); //边缘节点IP+端口号
ClusterConfig clusterConfigCom = new ClusterConfig();
clusterConfigCom.setId(11113l);
clusterConfigCom.setOwner("joliu");
List<ContainerDescription> cds = new ArrayList<>();
ContainerDescription containerDescriptionClient = new ContainerDescription();
containerDescriptionClient.setMode("normal");
containerDescriptionClient.setReplicas(7);
// satellite端
ContainerDescription containerDescriptionSatellite = new ContainerDescription();
containerDescriptionSatellite.setMode("normal");
containerDescriptionSatellite.setReplicas(3);
DockerContainer container1 = new DockerContainer();
container1.setName("sate");//创建satellite容器
container1.setCommand("bash");
container1.setImage("raft_sate");
containerDescriptionSatellite.setDockerContainer(container1);
cds.add(containerDescriptionSatellite);//将client容器添加到list集合中
// ground端
ContainerDescription containerDescriptionGround = new ContainerDescription();
containerDescriptionGround.setMode("normal");
containerDescriptionGround.setReplicas(1);
DockerContainer container = new DockerContainer();
container.setName("sate");//创建一个client容器
container.setName("ground");//创建一个ground容器
container.setCommand("bash");
container.setImage("raft_test");
containerDescriptionClient.setDockerContainer(container);
cds.add(containerDescriptionClient);//将client容器添加到list集合中
container.setImage("raft_ground");
containerDescriptionGround.setDockerContainer(container);
cds.add(containerDescriptionGround);//将client容器添加到list集合中
clusterConfigCom.setDockers(cds);
NetworkTopology topo = new NetworkTopology();
topo.setAppNames(new String[]{"sate", "br:ovs1"});
// topo.setAppNames(new String[]{"sate_0", "sate_1", "sate_2", "sate_3", "sate_4", "sate_5", "sate_6", "br:ovs1"});
// 这个参数好像没啥用
topo.setAppNames(new String[]{"sate", "ground", "br:ovs1"});
topo.setTopologyId(11);
// topo.setTopology(new int[][]{{0, 0, 0, 0, 0, 0, 0, 0},
// {0, 0, 0, 0, 0, 0, 0, 0},
// {0, 0, 0, 0, 0, 0, 0, 0},
// {0, 0, 0, 0, 0, 0, 0, 0},
// {0, 0, 0, 0, 0, 0, 0, 0},
// {0, 0, 0, 0, 0, 0, 0, 0},
// {0, 0, 0, 0, 0, 0, 0, 0},
// {1, 1, 1, 1, 1, 1, 1, 0}});
topo.setTopology(new int[][]{{0, 0}, {1, 0}});
topo.setTopology(new int[][]{{0,0,0}, {0,0,0}, {1,1,0}});
clusterConfigCom.setTopology(topo);
separatedClusterConfig.setClusterConfig(clusterConfigCom);
clusterConfigs.add(separatedClusterConfig);
......
cnf-test/src/test/java/top/ninwoo/YmwangDbcTests.java
View file @ 0e7f587b
......@@ -31,44 +31,59 @@ public class YmwangDbcTests {
ArrayList<SeparatedClusterConfig> clusterConfigs = new ArrayList<>();
SeparatedClusterConfig separatedClusterConfig = new SeparatedClusterConfig();
// TODO: 这个ID应该是从借口获取的
separatedClusterConfig.setEdgeNodeId("192.168.31.198:8881"); //边缘节点IP+端口号
separatedClusterConfig.setEdgeNodeId("192.168.31.213:18088"); //边缘节点IP+端口号
ClusterConfig clusterConfigCom = new ClusterConfig();
clusterConfigCom.setId(11113l);
clusterConfigCom.setOwner("joliu");
List<ContainerDescription> cds = new ArrayList<>();
/*通过镜像"ymwang/dfs_transfer_client"启动一个名为client的容器，来作为文件的收发端（图片的切割、合并）*/
ContainerDescription containerDescriptionClient = new ContainerDescription();
containerDescriptionClient.setMode("normal");
containerDescriptionClient.setReplicas(1);
DockerContainer container = new DockerContainer();
container.setName("dbc_client");//创建一个client容器
container.setCommand("bash");
container.setImage("dbc_client");
container.setPorts("8999");//指定容器端口
containerDescriptionClient.setDockerContainer(container);
cds.add(containerDescriptionClient);//将client容器添加到list集合中
/*通过镜像"ymwang/dfs_transfer_server"启动一个名为server1的容器，用于存储图片切片*/
ContainerDescription containerDescriptionServer1 = new ContainerDescription();
containerDescriptionServer1.setMode("normal");
containerDescriptionServer1.setReplicas(20);
DockerContainer containerServer1 = new DockerContainer();
containerServer1.setName("dbc_server");//创建一个server1容器
containerServer1.setCommand("sh");
containerServer1.setImage("dbc_server");
containerDescriptionServer1.setDockerContainer(containerServer1);
cds.add(containerDescriptionServer1);//将server1容器添加到list集合中
/*source节点*/
ContainerDescription containerDescriptionSource = new ContainerDescription();
containerDescriptionSource.setMode("normal");
containerDescriptionSource.setReplicas(1);
DockerContainer container1 = new DockerContainer();
container1.setName("ground0");//创建一个client容器
container1.setCommand("sh");
container1.setImage("source");
// container1.setPorts("2021");//指定容器端口
containerDescriptionSource.setDockerContainer(container1);
cds.add(containerDescriptionSource);//将client容器添加到list集合中
/*destination节点*/
ContainerDescription containerDescriptionDestination = new ContainerDescription();
containerDescriptionDestination.setMode("normal");
containerDescriptionDestination.setReplicas(1);
DockerContainer container2 = new DockerContainer();
container2.setName("ground7");//创建一个client容器
container2.setCommand("sh");
container2.setImage("destination");
// container2.setPorts("2021");//指定容器端口
containerDescriptionDestination.setDockerContainer(container2);
cds.add(containerDescriptionDestination);//将client容器添加到list集合中
/*satellite*/
ContainerDescription containerDescriptionSatellite = new ContainerDescription();
containerDescriptionSatellite.setMode("normal");
containerDescriptionSatellite.setReplicas(30);
DockerContainer container3 = new DockerContainer();
container3.setName("sate");//创建一个server1容器
container3.setCommand("sh");
container3.setImage("satellite");
// container3.setPorts("2021");//指定容器端口
containerDescriptionSatellite.setDockerContainer(container3);
cds.add(containerDescriptionSatellite);//将server1容器添加到list集合中
clusterConfigCom.setDockers(cds);
NetworkTopology topo = new NetworkTopology();
//topo.setAppNames(new String[]{"Run", "APP", "br:ovs1"});
topo.setAppNames(new String[]{"dbc_client", "dbc_server", "br:ovs1"});
topo.setAppNames(new String[]{"ground0","ground7", "sate", "br:ovs1", "br:ovs2"});
//topo.setAppNames(new String[]{"sate", "br:ovs1"});
// 这个参数好像没啥用
topo.setTopologyId(11);
topo.setTopology(new int[][]{{0,0,0},{0,0,0},{1,1,0}});
topo.setTopology(new int[][]{{0,0,0,0,0},{0,0,0,0,0},{0,0,0,0,0},{1,1,0,0,0},{0,1,1,0,0}});
//topo.setTopology(new int[][]{{0,0},{1,0}});
clusterConfigCom.setTopology(topo);
separatedClusterConfig.setClusterConfig(clusterConfigCom);
clusterConfigs.add(separatedClusterConfig);
......@@ -79,13 +94,13 @@ public class YmwangDbcTests {
//测试容器ip
@Test
public void getIpListByAppNameTest() {
List<String> ipList = networkService.getIpListByAppName(11113L, "dbc_server");
List<String> ipList = networkService.getIpListByAppName(11113L, "sate");
System.out.println(ipList);
}
@Test
public void getIpList() {
List<String> ipListS = networkService.getIpListByAppName(11113L, "dbc_server");
List<String> ipListS = networkService.getIpListByAppName(11113L, "sate");
String[] ipList = new String[ipListS.size()];
for (int i = 0; i < ipListS.size(); i++) {
if(!ipListS.isEmpty()){
......
cnf-utils/src/main/java/top/ninwoo/utils/util/impl/DockerUtilsImpl.java
View file @ 0e7f587b
......@@ -98,9 +98,9 @@ public class DockerUtilsImpl implements DockerUtils {
String cmd;
if(container.getPorts() == null){
// cmd = "docker run -itd --name " + container.getName() + " --privileged " + container.getImage() + " " + container.getCommand();
cmd = "docker run -itd -P " + " --name " + container.getName() + " --net=none" + " --privileged " + container.getImage() + " " + container.getCommand();
}else {
cmd = "docker run -itd -p " + container.getPorts()+ ":" +container.getPorts()+ " " + "--name " + container.getName() + " --net=none" + " --privileged " + container.getImage() + " " + container.getCommand();
cmd = "docker run -itd --name " + container.getName() + " --net=none" + " --privileged " + container.getImage() + " " + container.getCommand();
} else {
cmd = "docker run -itd -p " + container.getPorts() + ":" +container.getPorts()+ " " + "--name " + container.getName() + " --privileged " + container.getImage() + " " + container.getCommand();
}
String result = linuxCtlUtils.runCmd(cmd);
......
pom.xml
View file @ 0e7f587b
......@@ -19,7 +19,9 @@
<module>apps/cnf-weixingsim</module>
<module>apps/cnf-case-dis</module>
<module>apps/cnf-app-demo</module>
<module>apps/cnf-space-iot</module>
<module>apps/cnf-spaceiot-longConnection</module>
<module>apps/cnf-spaceiot-shortConnection</module>
<module>apps/cnf-spaceiot-mqtt</module>
</modules>
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment