CaiXiang/dc
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

完善 。S7 健康检查

Browse Source
This commit is contained in:
caixiang 2022-08-09 09:32:19 +08:00
parent 00a7aaaa37
commit 3b1c667915
9 changed files with 96 additions and 92 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
src/main/java/com/qgs/dc/influx/Main.java
View File

@ -136,7 +136,9 @@ public class Main {
// |> limit(n: 10, offset: 20) // |> limit(n: 10, offset: 20)
//22-57 22.3
//23-20 78
//24
// QueryDataParam queryDataParam = new QueryDataParam(); // QueryDataParam queryDataParam = new QueryDataParam();
// queryDataParam.setBucket("mytest"); // queryDataParam.setBucket("mytest");
// queryDataParam.setRange(new Range(getDate().toInstant(),new Date().toInstant())); // queryDataParam.setRange(new Range(getDate().toInstant(),new Date().toInstant()));

2
src/main/java/com/qgs/dc/influx/controller/InfluxDemoController.java
View File

@ -61,6 +61,8 @@ public class InfluxDemoController {
return point; return point;
} }
@PostMapping("/insert") @PostMapping("/insert")
public void insert() throws InterruptedException { public void insert() throws InterruptedException {
Event event = new Event(); Event event = new Event();

12
src/main/java/com/qgs/dc/s7/controller/S7DemoController.java
View File

@ -82,7 +82,7 @@ public class S7DemoController {
@PostMapping("/readTest2") @PostMapping("/readTest2")
public R getTest2ForS7() throws UnsupportedEncodingException, ParseException { public R getTest2ForS7() throws Exception {
// List<Character> characters = (List<Character>)read(PlcVarActual.CharArrays,S7Client.S7_1500); // List<Character> characters = (List<Character>)read(PlcVarActual.CharArrays,S7Client.S7_1500);
// //
// List<Boolean> booleans = (List<Boolean>)read(PlcVarActual.BooleanArrays,S7Client.S7_1500); // List<Boolean> booleans = (List<Boolean>)read(PlcVarActual.BooleanArrays,S7Client.S7_1500);
@ -99,7 +99,7 @@ public class S7DemoController {
} }
@PostMapping("/testForString") @PostMapping("/testForString")
public R testForStrings() throws UnsupportedEncodingException, ParseException { public R testForStrings() throws Exception {
//测试结果 l => 66ms //测试结果 l => 66ms
long l = System.currentTimeMillis(); long l = System.currentTimeMillis();
String[] subs = (String[])read(S7Client.S7_1500,PlcVarActual.SubIdArrays); String[] subs = (String[])read(S7Client.S7_1500,PlcVarActual.SubIdArrays);
@ -128,7 +128,7 @@ public class S7DemoController {
} }
@PostMapping("/testForString1200") @PostMapping("/testForString1200")
public R testForString1200() throws UnsupportedEncodingException, ParseException { public R testForString1200() throws Exception {
//测试结果 l => 66ms //测试结果 l => 66ms
long l = System.currentTimeMillis(); long l = System.currentTimeMillis();
String[] subs = (String[])read(S7Client.S7_1200,PlcVarActual.SubIdArrays1200); //65ms String[] subs = (String[])read(S7Client.S7_1200,PlcVarActual.SubIdArrays1200); //65ms
@ -151,7 +151,7 @@ public class S7DemoController {
return R.ok().put("l",(l1-l)).put("c",(c2-c1)); return R.ok().put("l",(l1-l)).put("c",(c2-c1));
} }
@PostMapping("/testForString1500") @PostMapping("/testForString1500")
public R testForString1500() throws UnsupportedEncodingException, ParseException { public R testForString1500() throws Exception {
//测试结果 l => 66ms //测试结果 l => 66ms
long l = System.currentTimeMillis(); long l = System.currentTimeMillis();
String[] subs = (String[])read(S7Client.S7_1500,PlcVarActual.SubIdArrays); //25ms String[] subs = (String[])read(S7Client.S7_1500,PlcVarActual.SubIdArrays); //25ms
@ -178,7 +178,7 @@ public class S7DemoController {
@PostMapping("/testFor1200") @PostMapping("/testFor1200")
public R testFor1200() throws UnsupportedEncodingException, ParseException { public R testFor1200() throws Exception {
//Object subs = read(PlcVarActual.INT1200, S7Client.S7_1200); //Object subs = read(PlcVarActual.INT1200, S7Client.S7_1200);
Object read = read(S7Client.S7_1200, PlcVarActual.SubIdArrays1200); Object read = read(S7Client.S7_1200, PlcVarActual.SubIdArrays1200);
@ -208,7 +208,7 @@ public class S7DemoController {
return null; return null;
} }
} }
private void write(S7Client s7Client,PlcVarActual var,Object newValue) { private void write(S7Client s7Client,PlcVarActual var,Object newValue) throws Exception {
if(var.getType().equals(PlcVar.STRING_Array)){ if(var.getType().equals(PlcVar.STRING_Array)){
String[] s = (String[])newValue; String[] s = (String[])newValue;
String[] ss = (String[])newValue; String[] ss = (String[])newValue;

2
src/main/java/com/qgs/dc/s7/my/s7connector/MainForRead.java
View File

@ -13,7 +13,7 @@ import java.util.List;
* @DATE: 2021/12/10 10:17 * @DATE: 2021/12/10 10:17
*/ */
public class MainForRead { public class MainForRead {
public static void main(String[] args) { public static void main(String[] args) throws Exception {
//前言： //前言：
//DB3.1.1 中间那个1是byte区，后面那个1 是bit //DB3.1.1 中间那个1是byte区，后面那个1 是bit
//缺陷： 不支持 DB3.1.1 //缺陷： 不支持 DB3.1.1

2
src/main/java/com/qgs/dc/s7/my/s7connector/MainForReadDemo.java
View File

@ -17,7 +17,7 @@ import java.util.List;
* @DATE: 2021/12/10 10:17 * @DATE: 2021/12/10 10:17
*/ */
public class MainForReadDemo { public class MainForReadDemo {
public static void main(String[] args) throws UnsupportedEncodingException { public static void main(String[] args) throws Exception {
// //1200 部分 // //1200 部分
// S7Connector connector1200 = S7Client.S7_1200.getConnector(); // S7Connector connector1200 = S7Client.S7_1200.getConnector();

2
src/main/java/com/qgs/dc/s7/my/s7connector/MainForWrite.java
View File

@ -18,7 +18,7 @@ import java.util.List;
* @DATE: 2021/12/10 10:17 * @DATE: 2021/12/10 10:17
*/ */
public class MainForWrite { public class MainForWrite {
public static void main(String[] args) throws IOException, ParseException { public static void main(String[] args) throws Exception {
//Create connection //Create connection
S7Connector connector = S7Connector connector =
S7ConnectorFactory S7ConnectorFactory

14
src/main/java/com/qgs/dc/s7/my/s7connector/api/S7Connector.java
View File

@ -23,36 +23,36 @@ import java.io.Closeable;
public interface S7Connector extends Closeable { public interface S7Connector extends Closeable {
/** /**
* Reads an area * Reads an area
* * desc : 只要未抛出异常，都是 操作成功的
* @param area * @param area
* @param areaNumber * @param areaNumber
* @param bytes * @param bytes
* @param offset * @param offset
* @return * @return
*/ */
public byte[] read(DaveArea area, int areaNumber, int bytes, int offset); public byte[] read(DaveArea area, int areaNumber, int bytes, int offset) throws Exception;
/** /**
* Reads an area 读需要bit 位置的 变量（其实就是 read bool变量的时候调用这个方法。） * Reads an area 读需要bit 位置的 变量（其实就是 read bool变量的时候调用这个方法。）
* * desc : 只要未抛出异常，都是 操作成功的
* @param area * @param area
* @param areaNumber * @param areaNumber
* @param bytes * @param bytes
* @param offset * @param offset
* @return * @return
*/ */
public byte[] read(DaveArea area, int areaNumber, int bytes, int offset, int bitOffset, TransportSize transportSize); public byte[] read(DaveArea area, int areaNumber, int bytes, int offset, int bitOffset, TransportSize transportSize) throws Exception;
/** /**
* Writes an area * Writes an area
* * desc : 只要未抛出异常，都是 操作成功的
* @param area * @param area
* @param areaNumber * @param areaNumber
* @param offset * @param offset
* @param buffer * @param buffer
*/ */
public void write(DaveArea area, int areaNumber, int offset, byte[] buffer); public void write(DaveArea area, int areaNumber, int offset, byte[] buffer) throws Exception;
//如果 bitOffset 没有 那么就填0 //如果 bitOffset 没有 那么就填0
public void write(DaveArea area, int areaNumber, int byteOffset, int bitOffset, byte[] buffer, PlcVar var); public void write(DaveArea area, int areaNumber, int byteOffset, int bitOffset, byte[] buffer, PlcVar var) throws Exception;
} }

137
src/main/java/com/qgs/dc/s7/my/s7connector/enmuc/S7Client.java
View File

@ -16,6 +16,8 @@ import java.util.Arrays;
import java.util.List; import java.util.List;
import java.util.concurrent.Executors; import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
/** /**
* @Desc: "" * @Desc: ""
@ -114,7 +116,7 @@ public enum S7Client {
* *
* *
* */ * */
public Object read(DaveArea area, Integer areaNumber, Integer byteOffset, Integer bitOffset, Integer length, Integer strSizes, PlcVar type) throws UnsupportedEncodingException, ParseException { public Object read(DaveArea area, Integer areaNumber, Integer byteOffset, Integer bitOffset, Integer length, Integer strSizes, PlcVar type) throws Exception {
S7Connector connector = getConnector(); S7Connector connector = getConnector();
//String 类型比较特殊。 String[] 也是同理，Sring数组里面的子项 也是有两个字节的 readBytes。 //String 类型比较特殊。 String[] 也是同理，Sring数组里面的子项 也是有两个字节的 readBytes。
if(type.equals(PlcVar.STRING)){ if(type.equals(PlcVar.STRING)){
@ -212,7 +214,7 @@ public enum S7Client {
* *
* *
* */ * */
public void write(DaveArea area, Integer areaNumber, Integer byteOffset, Integer bitOffset,Integer strSize, PlcVar type, Object newValue){ public void write(DaveArea area, Integer areaNumber, Integer byteOffset, Integer bitOffset,Integer strSize, PlcVar type, Object newValue) throws Exception {
S7Connector connector = getConnector(); S7Connector connector = getConnector();
//String 类型比较特殊。 String[] 也是同理，Sring数组里面的子项 也是有两个字节的 readBytes。 //String 类型比较特殊。 String[] 也是同理，Sring数组里面的子项 也是有两个字节的 readBytes。
@ -281,89 +283,86 @@ public enum S7Client {
} }
private void check_ping(){ private void check_ping(){
ping_fail_check.execute(new Runnable() { while (true){
@Override try {
public void run() { ScheduledFuture<?> schedule = ping_fail_check.schedule(new Runnable() {
while (true){ @Override
if(connections.size()!=coreSize){ public void run() {
int c = coreSize-connections.size(); if (connections.size() != coreSize) {
for(int z=0;z<c;z++){ int c = coreSize - connections.size();
S7Connector connect = connect(host, rack, slot); for (int z = 0; z < c; z++) {
if(connect!=null){ S7Connector connect = connect(host, rack, slot);
logger.info("host:"+host +" ;;(check_ping()) 检测到有断线 ==》 现在断线恢复成功"); if (connect != null) {
connections.add(connect); logger.info("host:" + host + " ;;(check_ping()) 检测到有断线 ==》 现在断线恢复成功");
}else { connections.add(connect);
logger.info("host:"+host +" ;;(check_ping()) 检测到有断线 ==》 现在断线恢复失败"); } else {
} logger.info("host:" + host + " ;;(check_ping()) 检测到有断线 ==》 现在断线恢复失败");
}
}
} }
} }
try { }, 10000, TimeUnit.SECONDS);
Thread.sleep(10000); Object o = schedule.get();
}catch (Exception e){ }catch (Exception e){
logger.info("host:"+host +" ;;(check_ping()) Thread.sleep 异常，异常原因："+e.getMessage()); logger.info("( out catched ) host:"+host +" ;;(check_ping) "+"check connect.size 出现异常，errMessage is : "+e.getMessage()+"现在进入下一次轮询");
}
}
} }
}); }
} }
//todo 当网络断开之后，下面这个ping() 循环不生效。。 可能是read = connector.read( 出来是null，但还是继续通过了，也可能是 ，，当连接全部断开后 getConnector(); 应该是取不到数据的。。也就会报null。这个明天再看看 //todo 当网络断开之后，下面这个ping() 循环不生效。。 可能是read = connector.read( 出来是null，但还是继续通过了，也可能是 ，，当连接全部断开后 getConnector(); 应该是取不到数据的。。也就会报null。这个明天再看看
private void ping(){ private void ping(){
executor.execute(new Runnable() { while (true){
@Override try {
public void run() { ScheduledFuture<?> schedule = executor.schedule(new Runnable() {
while (true){ @Override
for(int i=0;i<coreSize;i++){ public void run() {
S7Connector connector = getConnector(); for (int i = 0; i < coreSize; i++) {
try { S7Connector connector = getConnector();
//只要没有报异常 都是通讯正常的。
byte[] read = connector.read(
heartBeat.getArea(),
heartBeat.getAreaNumber(),
heartBeat.getType().getTransportSize().getSizeInBytes(),
heartBeat.getByteOffset(),
heartBeat.getBitOffset(),
heartBeat.getType().getTransportSize());
System.out.println("host:"+host +" ;; "+connector.hashCode()+" : ping");
Thread.sleep(100);
}catch (Exception e){
logger.info("host:"+host +" ;;(ping) "+connector.hashCode()+" : connection error"+"errMessage is : "+e.getMessage());
//先把 socket close掉
try { try {
connector.close(); //只要没有报异常 都是通讯正常的。
connections.remove(connector); byte[] read = connector.read(
//如果是网络波动照成的socket断开。 等个1S 再重连试试 heartBeat.getArea(),
heartBeat.getAreaNumber(),
heartBeat.getType().getTransportSize().getSizeInBytes(),
heartBeat.getByteOffset(),
heartBeat.getBitOffset(),
heartBeat.getType().getTransportSize());
System.out.println("host:" + host + " ;; " + connector.hashCode() + " : ping");
Thread.sleep(100); Thread.sleep(100);
}catch (Exception ee){ } catch (Exception e) {
logger.info("host:"+host +" ;;(ping) "+"connector.close() 出现异常，errMessage is : "+ee.getMessage()); logger.info("host:" + host + " ;;(ping) " + connector.hashCode() + " : connection error" + "errMessage is : " + e.getMessage());
//先把 socket close掉
try {
connector.close();
connections.remove(connector);
//如果是网络波动照成的socket断开。 等个1S 再重连试试
Thread.sleep(100);
} catch (Exception ee) {
logger.info("host:" + host + " ;;(ping) " + "connector.close() 出现异常，errMessage is : " + ee.getMessage());
}
} }
}
//todo 把之前的连接close 掉，然后新增一个连接到connections //todo 把之前的连接close 掉，然后新增一个连接到connections
if(connections.size()!=coreSize){ if (connections.size() != coreSize) {
int c = coreSize-connections.size(); int c = coreSize - connections.size();
for(int z=0;z<c;z++){ for (int z = 0; z < c; z++) {
S7Connector connect = connect(host, rack, slot); S7Connector connect = connect(host, rack, slot);
if(connect!=null){ if (connect != null) {
logger.info("host:"+host +" ;;(ping) "+"ping时候出现异常，尝试重连， 重连成功!!"); logger.info("host:" + host + " ;;(ping) " + "ping时候出现异常，尝试重连， 重连成功!!");
connections.add(connect); connections.add(connect);
}else { } else {
logger.info("host:"+host +" ;;(ping) "+"ping时候出现异常，尝试重连， 重连时候还是 出现异常。。"); logger.info("host:" + host + " ;;(ping) " + "ping时候出现异常，尝试重连， 重连时候还是 出现异常。。");
}
} }
} }
} }
} }
}, 30000, TimeUnit.SECONDS);
try { Object o = schedule.get();
Thread.sleep(30000); }catch (Exception e){
}catch (Exception e){ logger.info("( out catched ) host:"+host +" ;;(ping) "+"ping的时候 整体 出现异常，errMessage is : "+e.getMessage()+"现在进入下一次轮询");
e.printStackTrace();
logger.info("host:"+host +" ;;(ping) "+e.getMessage());
}
}
} }
}); }
} }
} }

13
src/main/java/com/qgs/dc/s7/my/s7connector/impl/S7BaseConnection.java
View File

@ -52,10 +52,11 @@ public abstract class S7BaseConnection implements S7Connector {
* @param libnodaveResult * @param libnodaveResult
* the libnodave result * the libnodave result
*/ */
public static void checkResult(final int libnodaveResult) { public static void checkResult(final int libnodaveResult) throws Exception {
if (libnodaveResult != Nodave.RESULT_OK) { if (libnodaveResult != Nodave.RESULT_OK) {
final String msg = Nodave.strerror(libnodaveResult); final String msg = Nodave.strerror(libnodaveResult);
throw new IllegalArgumentException("Result: " + msg); // throw new IllegalArgumentException("Result: " + msg);
throw new Exception(msg);
} }
} }
@ -86,7 +87,7 @@ public abstract class S7BaseConnection implements S7Connector {
/** {@inheritDoc} */ /** {@inheritDoc} */
@Override @Override
public synchronized byte[] read(final DaveArea area, final int areaNumber, final int bytes, final int offset) { public synchronized byte[] read(final DaveArea area, final int areaNumber, final int bytes, final int offset) throws Exception {
if (bytes > MAX_SIZE) { if (bytes > MAX_SIZE) {
final byte[] ret = new byte[bytes]; final byte[] ret = new byte[bytes];
//注意这里 嵌套了 递归，让无限递归去解决 MAX_SIZE的问题。 //注意这里 嵌套了 递归，让无限递归去解决 MAX_SIZE的问题。
@ -107,7 +108,7 @@ public abstract class S7BaseConnection implements S7Connector {
} }
} }
@Override @Override
public synchronized byte[] read(final DaveArea area, final int areaNumber, final int bytes, final int offset, int bitOffset, TransportSize transportSize) { public synchronized byte[] read(final DaveArea area, final int areaNumber, final int bytes, final int offset, int bitOffset, TransportSize transportSize) throws Exception {
if(bitOffset==0){ if(bitOffset==0){
return read(area,areaNumber,bytes,offset); return read(area,areaNumber,bytes,offset);
} }
@ -134,7 +135,7 @@ public abstract class S7BaseConnection implements S7Connector {
/** {@inheritDoc} */ /** {@inheritDoc} */
@Override @Override
public synchronized void write(final DaveArea area, final int areaNumber, final int offset, final byte[] buffer) { public synchronized void write(final DaveArea area, final int areaNumber, final int offset, final byte[] buffer) throws Exception {
if (buffer.length > MAX_SIZE) { if (buffer.length > MAX_SIZE) {
// Split buffer // Split buffer
final byte[] subBuffer = new byte[MAX_SIZE]; final byte[] subBuffer = new byte[MAX_SIZE];
@ -154,7 +155,7 @@ public abstract class S7BaseConnection implements S7Connector {
} }
@Override @Override
public synchronized void write(final DaveArea area, final int areaNumber, final int byteOffset, final int bitOffset, final byte[] buffer, PlcVar var) { public synchronized void write(final DaveArea area, final int areaNumber, final int byteOffset, final int bitOffset, final byte[] buffer, PlcVar var) throws Exception {
if(bitOffset==0){ if(bitOffset==0){
write(area,areaNumber,byteOffset,buffer); write(area,areaNumber,byteOffset,buffer);
return; return;