Imdroid/security-monitor
2
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

1、修改TCP推送的问题

Browse Source
This commit is contained in:
weidong 2024-01-31 07:56:59 +08:00
parent 1bb6946bf8
commit 3a7350f0f7
5 changed files with 37 additions and 51 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
sec-beidou-fwd/src/main/java/com/imdroid/beidou_fwd/service/TCPClient.java
View File

@ -1,5 +1,6 @@
package com.imdroid.beidou_fwd.service;
import com.imdroid.common.util.ThreadManager;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
@ -11,6 +12,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.nio.charset.StandardCharsets;
import java.time.LocalDateTime;
import java.util.concurrent.TimeUnit;
public class TCPClient {
@ -25,17 +27,15 @@ public class TCPClient {
private Bootstrap bootstrap;
private EventLoopGroup group;
private Channel channel;
ByteBuf sendBuffer;
TCPListener listener;
LocalDateTime connectTime = LocalDateTime.now();
public void start() {
new Thread(this::connect, "xfz-tcp-client").start();
}
public void init(String dest_addr, int dest_port, TCPListener listener) {
public void init(String dest_addr, int dest_port) {
this.host = dest_addr;
this.port = dest_port;
this.listener = listener;
//客户端需要一个事件循环组
group = new NioEventLoopGroup();
@ -85,33 +85,34 @@ public class TCPClient {
}
public void writeAndFlush(String json) {
sendBuffer = Unpooled.buffer();
ByteBuf sendBuffer = Unpooled.buffer();
sendBuffer.writeBytes(json.getBytes(StandardCharsets.UTF_8));
if(channel.isActive()) flush();
else connect();
channel.writeAndFlush(sendBuffer).addListener(future -> {
if (future.isSuccess()) {
logger.info("send to xfz server succeed.");
} else {
logger.info("send to xfz server failed.");
}
});
}
public void onConnected(){
listener.onConnected();
flush();
connectTime = LocalDateTime.now();
}
public void onDisconnect(){
listener.onDisconnect();
}
private void flush(){
if(sendBuffer!=null && sendBuffer.readableBytes()>0){
channel.writeAndFlush(sendBuffer).addListener(future -> {
if (future.isSuccess()) {
logger.info("send to xfz server succeed.");
//sendBuffer.release(); // writeAndFlush后就释放了
listener.onSendResult(true);
} else {
logger.info("send to xfz server failed.");
listener.onSendResult(false);
if(connectTime.isBefore(LocalDateTime.now().minusMinutes(1))) {
connect();
}
else{
ThreadManager.getScheduledThreadPool().schedule(() -> {
try {
connect();
} catch (Exception e) {
logger.error(e.toString());
}
});
},60, TimeUnit.SECONDS);
}
}
}

26
sec-beidou-fwd/src/main/java/com/imdroid/beidou_fwd/task/GXXfzForwarder.java
View File

@ -2,7 +2,6 @@ package com.imdroid.beidou_fwd.task;
import com.imdroid.beidou_fwd.entity.XFZData;
import com.imdroid.beidou_fwd.service.TCPClient;
import com.imdroid.beidou_fwd.service.TCPListener;
import com.imdroid.common.util.GsonUtil;
import com.imdroid.common.util.NumberUtils;
import com.imdroid.secapi.dto.*;
@ -36,41 +35,24 @@ public class GXXfzForwarder extends Forwarder{
final DateTimeFormatter dateFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
static class GXXfzTCPListener implements TCPListener{
@Override
public void onConnected() {
}
@Override
public void onDisconnect() {
}
@Override
public void onSendResult(boolean isOK) {
}
}
@PostConstruct
void registerMe(){
init(FORWARDER_NAME, "TCP "+host+":"+port,1,false);
xfzTcpClient = new TCPClient();
xfzTcpClient.init(host, port, new GXXfzTCPListener());
xfzTcpClient.init(host, port);
xfzTcpClient.start();
}
/**
* 每半小时转发GNSS解算结果
*/
@Scheduled(cron = "0 0,30 * * * ?") // 每30分钟执行一次
@Scheduled(cron = "0 0/30 * * * ?") // 每30分钟执行一次
private void forwardGnss() {
logger.info("xfz forwardGnss");
forwardCurrentGnss(FORWARDER_NAME);
}
@Scheduled(cron = "0 38 * * * ?") // 每小时的40分钟执行一次
@Scheduled(cron = "0 40 * * * ?") // 每小时的40分钟执行一次
//@Scheduled(cron = "0 0/20 * * * ?") // 每20分钟执行一次
private void forwardHistoryGnss() {
forwardHistoryGnss(FORWARDER_NAME);

5
sec-beidou-fwd/src/main/java/com/imdroid/beidou_fwd/task/GZYForwarder.java
View File

@ -40,12 +40,13 @@ public class GZYForwarder extends Forwarder{
/**
* 每半小时转发GNSS解算结果
*/
@Scheduled(cron = "0 10,40 * * * ?") // 每30分钟执行一次
@Scheduled(cron = "0 0/30 * * * ?") // 每30分钟执行一次
private void forwardGnss() {
logger.info("gzy UDP forwardGnss");
forwardCurrentGnss(FORWARDER_NAME);
}
@Scheduled(cron = "0 48 * * * ?") // 每小时的40分钟执行一次
@Scheduled(cron = "0 40 * * * ?") // 每小时的40分钟执行一次
//@Scheduled(cron = "0 0/20 * * * ?") // 每20分钟执行一次
private void forwardHistoryGnss() {
forwardHistoryGnss(FORWARDER_NAME);

5
sec-beidou-fwd/src/main/java/com/imdroid/beidou_fwd/task/GZYMQTTForwarder.java
View File

@ -49,12 +49,13 @@ public class GZYMQTTForwarder extends Forwarder {
/**
* 每半小时转发GNSS解算结果
*/
@Scheduled(cron = "0 10,40 * * * ?") // 每30分钟执行一次
@Scheduled(cron = "0 0/30 * * * ?") // 每30分钟执行一次
private void forwardGnss() {
logger.info("gzy mqtt forwardGnss");
forwardCurrentGnss(FORWARDER_NAME);
}
@Scheduled(cron = "0 48 * * * ?") // 每小时的40分钟执行一次
@Scheduled(cron = "0 40 * * * ?") // 每小时的40分钟执行一次
//@Scheduled(cron = "0 0/20 * * * ?") // 每20分钟执行一次
private void forwardHistoryGnss() {
forwardHistoryGnss(FORWARDER_NAME);

5
sec-beidou-fwd/src/main/java/com/imdroid/beidou_fwd/task/KingMaForwarder.java
View File

@ -48,12 +48,13 @@ public class KingMaForwarder extends Forwarder{
/**
* 每半小时转发GNSS解算结果
*/
@Scheduled(cron = "0 20,50 * * * ?") // 每30分钟执行一次
@Scheduled(cron = "0 0/30 * * * ?") // 每30分钟执行一次
private void forwardGnss() {
logger.info("kingma forwardGnss");
forwardCurrentGnss(FORWARDER_NAME);
}
@Scheduled(cron = "0 58 * * * ?") // 每小时的40分钟执行一次
@Scheduled(cron = "0 40 * * * ?") // 每小时的40分钟执行一次
//@Scheduled(cron = "0 0/20 * * * ?") // 每20分钟执行一次
private void forwardHistoryGnss() {
forwardHistoryGnss(FORWARDER_NAME);