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-16 08:13:21 +08:00
parent 45fa9509e7
commit ae1d9c8287
3 changed files with 31 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -22,10 +22,10 @@ public class TCPClient {
//@Value("${xfz.server.port}")
private int port;
private Bootstrap bootstrap;
private EventLoopGroup group;
private Channel channel;
ByteBuf sendBuffer;
public void start() {
new Thread(this::connect, "xfz-tcp-client").start();
@ -34,6 +34,7 @@ public class TCPClient {
public void init(String dest_addr, int dest_port) {
host = dest_addr;
port = dest_port;
sendBuffer = Unpooled.buffer();
//客户端需要一个事件循环组
group = new NioEventLoopGroup();
//创建客户端启动对象
@ -45,7 +46,7 @@ public class TCPClient {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
//加入处理器
ch.pipeline().addLast(new TcpMessageHandler());
ch.pipeline().addLast(new TcpMessageHandler(TCPClient.this));
}
});
}
@ -81,21 +82,26 @@ public class TCPClient {
}
}
public void writeAndFlush(String json) {
String str = "#" + json + "!";
ByteBuf buf = Unpooled.buffer();
buf.writeBytes(str.getBytes(StandardCharsets.UTF_8));
channel.writeAndFlush(buf).addListener(future -> {
if (future.isSuccess()) {
logger.info("send {} to xfz server succeed.", str);
} else {
logger.info("send {} to xfz server failed.", str);
}
});
sendBuffer.writeBytes(json.getBytes(StandardCharsets.UTF_8));
if(channel.isActive()) flush();
else connect();
}
public boolean isActive(){
return channel.isActive();
public void onConnectionActive(){
flush();
}
private void flush(){
if(sendBuffer.readableBytes()>0){
channel.writeAndFlush(sendBuffer).addListener(future -> {
if (future.isSuccess()) {
logger.info("send to xfz server succeed.");
sendBuffer.clear();
} else {
logger.info("send to xfz server failed.");
}
});
}
}
}

7
sec-beidou-fwd/src/main/java/com/imdroid/beidou_fwd/service/TcpMessageHandler.java
View File

@ -15,6 +15,12 @@ public class TcpMessageHandler extends SimpleChannelInboundHandler<ByteBuf> {
private final Logger logger = LoggerFactory.getLogger(this.getClass());
private final TCPClient tcpClient;
public TcpMessageHandler(TCPClient tcpClient) {
this.tcpClient = tcpClient;
}
@Override
protected void channelRead0(ChannelHandlerContext channelHandlerContext, ByteBuf buf) throws Exception {
if (logger.isDebugEnabled()) {
@ -27,6 +33,7 @@ public class TcpMessageHandler extends SimpleChannelInboundHandler<ByteBuf> {
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
logger.info("xfz tcp channel active");
tcpClient.onConnectionActive();
}
@Override

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

@ -82,26 +82,11 @@ public class GXXfzForwarder extends Forwarder{
data.setZ(NumberUtils.scale(locationRecord.getRb562d() * 0.001, 5));
sendNum++;
}
String json = GsonUtil.toJson(xfzTcpMessage);
String json = "#" + GsonUtil.toJson(xfzTcpMessage) + "!";
logger.info("project " + projectId + ": push calculation result to XFZ");
logger.info(json);
if(!xfzTcpClient.isActive()) {
xfzTcpClient.connect();
try {
Thread.sleep(5000);
}
catch (Exception e){
}
}
if(xfzTcpClient.isActive()) {
xfzTcpClient.writeAndFlush(json);
return sendNum;
}
else{
logger.info("TCP disconnected!");
return 0;
}
xfzTcpClient.writeAndFlush(json);
return sendNum;
}
}