Skip to content

T
traffic-signal-platform
Commit 7439d1d4 authored by duanruiming's avatar duanruiming
Browse files
Options

[add] 代码优化

parent 760d8b20
Hide whitespace changes
Inline Side-by-side
Showing with 118 additions and 149 deletions
signal-utc-dt-service/src/main/java/net/wanji/com/Application.java
View file @ 7439d1d4
......@@ -5,8 +5,7 @@ import org.mybatis.spring.annotation.MapperScan;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableScheduling;
......@@ -22,22 +21,21 @@ import org.springframework.transaction.annotation.EnableTransactionManagement;
@EnableTransactionManagement
@EnableScheduling
@SuppressWarnings("all")
public class Application implements ApplicationRunner {
public class Application implements CommandLineRunner {
@Qualifier(value = "threadPoolExecutor")
@Autowired
static ThreadPoolTaskExecutor threadPoolExecutor;
ThreadPoolTaskExecutor threadPoolExecutor;
@Value("${portParam.localPort}")
int localPort;
@Value("${portParam.remotePort}")
int remoteProt;
public static void main(String[] args) {
SpringApplication.run(Application.class, args);
}
@Override
public void run(ApplicationArguments args) throws Exception {
public void run(String... args) throws Exception {
NettyClient.start(localPort, remoteProt, threadPoolExecutor);
}
}
signal-utc-dt-service/src/main/java/net/wanji/com/cache/CrossInfoCache.java
View file @ 7439d1d4
......@@ -4,7 +4,8 @@ import lombok.extern.slf4j.Slf4j;
import net.wanji.databus.dao.mapper.CrossInfoMapper;
import net.wanji.databus.po.CrossInfoPO;
import org.apache.commons.lang3.StringUtils;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
......@@ -20,7 +21,7 @@ import java.util.Map;
@Slf4j
@Component
@SuppressWarnings("unchecked")
public class CrossInfoCache implements CommandLineRunner {
public class CrossInfoCache implements ApplicationRunner {
@Resource
private CrossInfoMapper crossInfoMapper;
......@@ -42,7 +43,7 @@ public class CrossInfoCache implements CommandLineRunner {
}
@Override
public void run(String... args) throws Exception {
public void run(ApplicationArguments args) throws Exception {
init();
}
......
signal-utc-dt-service/src/main/java/net/wanji/com/common/constants/Constants.java
View file @ 7439d1d4
......@@ -10,12 +10,10 @@ import java.util.List;
@SuppressWarnings("all")
public class Constants {
public static final String COMMAND_HEAD = "0x7e";
public static final String COMMAND_TAIL = "0x7d";
public static final String COMMAND_HEAD = "7e";
public static final String COMMAND_TAIL = "7d";
public static final String COMMAND_HEARTBEAT = "01000100000001000170";
public static final String COMMAND_SET_IP = "";
/**
* 消息体添加头、尾
* @param body
......
signal-utc-dt-service/src/main/java/net/wanji/com/common/enums/CommandResultSign.java
View file @ 7439d1d4
......@@ -10,7 +10,7 @@ import org.apache.commons.lang3.StringUtils;
@Getter
public enum CommandResultSign {
HEARTBEAT("80", "heartBeatService"),
HEARTBEAT("55ab", "heartBeatService"),
GET_DEVICE_INFO("2001", "getDeviceInfoService"),
GET_SIGNAL_BASE_INFO("2002","getBaseInfoService"),
GET_LIGHTS_GROUP_INFO("2003", "getLightsGroupInfoService"),
......
signal-utc-dt-service/src/main/java/net/wanji/com/config/ThreadPoolConfig.java
View file @ 7439d1d4
......@@ -28,7 +28,7 @@ public class ThreadPoolConfig {
@Value("${threadPoolConfig.allowCoreTimeOut}")
private boolean allowCoreTimeOut;
@Bean
@Bean("threadPoolExecutor")
public ThreadPoolTaskExecutor threadPoolExecutor() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
ThreadFactory build = new ThreadFactoryBuilder().setNameFormat(threadPoolName).build();
......
signal-utc-dt-service/src/main/java/net/wanji/com/netty/NettyClient.java
View file @ 7439d1d4
package net.wanji.com.netty;
import cn.hutool.core.util.HexUtil;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.util.concurrent.GlobalEventExecutor;
import io.netty.channel.socket.DatagramPacket;
import io.netty.channel.socket.nio.NioDatagramChannel;
import lombok.extern.slf4j.Slf4j;
import net.wanji.com.cache.netty.NettyMessageCache;
import net.wanji.com.netty.codec.MessageDecoder;
import net.wanji.com.netty.codec.MessageEnCoder;
import net.wanji.com.netty.handler.NettyServerHandler;
import net.wanji.com.pojo.netty.MessageResultPojo;
import org.apache.commons.lang3.StringUtils;
......@@ -19,7 +20,6 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;
import java.net.InetSocketAddress;
import java.nio.ByteOrder;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.HashMap;
......@@ -32,7 +32,7 @@ import java.util.concurrent.TimeUnit;
public class NettyClient {
private static final Map<String, ChannelId> IP_PORT_CHANNEL_ID_MAP = new HashMap<>();
private static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
private static ChannelFuture udpChannelFuture = null;
/**
* 建立连接
......@@ -41,52 +41,49 @@ public class NettyClient {
EventLoopGroup group = new NioEventLoopGroup();
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(group)
.channel(NioSocketChannel.class)
.option(ChannelOption.TCP_NODELAY, true)
.option(ChannelOption.SO_KEEPALIVE, true)
.channel(NioDatagramChannel.class)
.option(ChannelOption.RCVBUF_ALLOCATOR, new FixedRecvByteBufAllocator(65535))
// .localAddress(new InetSocketAddress(5051))
// 绑定端口
.remoteAddress(new InetSocketAddress(remotePort))
.handler(new ChannelInitializer<NioSocketChannel>() {
.handler(new ChannelInitializer<NioDatagramChannel>() {
@Override
protected void initChannel(NioSocketChannel socketChannel) {
log.info("连接channel:{}", socketChannel);
socketChannel.pipeline()
.addFirst()
.addLast(new LengthFieldBasedFrameDecoder(ByteOrder.LITTLE_ENDIAN, Integer.MAX_VALUE, 4, 2, 0, 0, true))
.addLast(new NettyServerHandler(threadPoolExecutor));
channelGroup.add(socketChannel);
socketChannel.writeAndFlush("2a320d0a24340d0a415554480d0a2431360d0a536d2a674234504742333939514241390d0a");
socketChannel.writeAndFlush("2a320d0a24340d0a415554480d0a2431360d0a536d2a674234504742333939514241390d0a2a330d0a24360d0a434c49454e540d0a24370d0a5345544e414d450d0a2431310d0a574a2d31303530333036320d0a2a320d0a24340d0a4543484f0d0a2431360d0a5135ee72dcef1143abbebbd4de9e5cbf0d0a2a320d0a24390d0a5355425343524942450d0a2432360d0a5f5f426f6f6b736c656576655f4d61737465724368616e6765640d0a");
log.info("连接成功");
protected void initChannel(NioDatagramChannel datagramChannel) {
datagramChannel.pipeline()
.addLast("decoder", new MessageDecoder())
.addLast("encoder", new MessageEnCoder())
.addLast(new NettyServerHandler());
}
});
ChannelFuture channelFuture = bootstrap.bind(localPort).sync();
Channel channel = channelFuture.channel();
channel.writeAndFlush("2a320d0a24340d0a415554480d0a2431360d0a536d2a674234504742333939514241390d0a");
log.info("server is started in {}", channelFuture.channel().localAddress());
//收到停止信号
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
//停掉所有的线程
group.shutdownGracefully();
channelFuture.channel().closeFuture().syncUninterruptibly();
log.error("server is closed");
log.warn("server is closed");
}));
log.info("udp application is running. binding port is {}", remotePort);
NettyClient.udpChannelFuture = channelFuture;
}
public static void sendMessage(String ip, int port, String message) {
Channel channel = channelGroup.find(IP_PORT_CHANNEL_ID_MAP.get(StringUtils.join(ip, ":", port)));
channel.writeAndFlush(message);
}
public static Object sendMessage(String ip, int port, String message, String command) {
sendMessage(ip, port, message, command, 500);
return null;
public static void sendMessage(String ip, Integer port, String msg) {
byte[] bytes = HexUtil.decodeHex(msg);
try {
if (udpChannelFuture == null) {
return;
}
udpChannelFuture.channel()
.writeAndFlush(new DatagramPacket(Unpooled.copiedBuffer(bytes), new InetSocketAddress(ip, port))).sync();
log.info("send msg {} to {}/{}", msg, ip, port);
} catch (InterruptedException e) {
log.error("sendMsg is error", e);
}
}
public static MessageResultPojo sendMessage(String ip, Integer port, String msg, String command, int timeout) {
try {
if (timeout > 0) {
CountDownLatch countDownLatch = new CountDownLatch(1);
String key = StringUtils.joinWith("#", ip, port, command);
String key = StringUtils.join("/", ip, ":", port, "/", command);
long now = LocalDateTime.now().toInstant(ZoneOffset.of("+8")).toEpochMilli();
MessageResultPojo messageResultPojo = new MessageResultPojo(command, countDownLatch, now, timeout, null);
NettyMessageCache.NETTY_MESSAGE_RESULT_MAP.put(key, messageResultPojo);
......
signal-utc-dt-service/src/main/java/net/wanji/com/netty/agreementcommon/AgreementCommon.java → signal-utc-dt-service/src/main/java/net/wanji/com/netty/agreementcommon/CommandResultSign.java
View file @ 7439d1d4
package net.wanji.com.netty.agreementcommon;
import java.util.Objects;
import net.wanji.common.utils.tool.StringUtils;
/**
* @author duanruiming
* @date 2023/05/08 10:14
*/
public class AgreementCommon {
public class CommandResultSign {
public static String getHexSign(String data) {
//根据不同需求获取消息唯一标识
String controlHex = data.substring(28, 30);
// 控制类型
if (Objects.equals("11", controlHex)) {
return "11";
} else {
String hex = data.substring(28, 32);
return hex;
String sign = data.substring(0, 4);
if (StringUtils.equalsIgnoreCase(sign, "")) {
}
return sign;
}
}
signal-utc-dt-service/src/main/java/net/wanji/com/netty/codec/MessageDecoder.java
View file @ 7439d1d4
package net.wanji.com.netty.codec;
import cn.hutool.core.util.HexUtil;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
......@@ -16,6 +17,8 @@ public class MessageDecoder extends ByteToMessageDecoder {
*/
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
byte[] bytes = new byte[in.readableBytes()];
in.readBytes(bytes);
out.add(HexUtil.encodeHexStr(bytes));
}
}
signal-utc-dt-service/src/main/java/net/wanji/com/netty/handler/NettyServerHandler.java
View file @ 7439d1d4
package net.wanji.com.netty.handler;
import cn.hutool.core.util.HexUtil;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.socket.DatagramPacket;
import lombok.extern.slf4j.Slf4j;
import net.wanji.com.cache.CrossInfoCache;
import net.wanji.com.cache.netty.NettyMessageCache;
import net.wanji.com.common.enums.CommandResultSign;
import net.wanji.com.netty.agreementcommon.AgreementCommon;
import net.wanji.com.netty.agreementcommon.CommandResultSign;
import net.wanji.com.netty.pojo.CommandPojo;
import net.wanji.com.netty.response.CommandResponseService;
import net.wanji.com.pojo.netty.MessageResultPojo;
import net.wanji.common.framework.spring.ServiceBeanContext;
import net.wanji.databus.po.CrossInfoPO;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.net.InetSocketAddress;
import java.util.Objects;
@Slf4j
public class NettyServerHandler extends ChannelInboundHandlerAdapter {
import static net.wanji.com.common.enums.CommandResultSign.getClassNameByHexSign;
@Resource
private CrossInfoCache crossInfoCache;
private ThreadPoolTaskExecutor threadPoolExecutor;
@Slf4j
@Component
public class NettyServerHandler extends SimpleChannelInboundHandler<DatagramPacket> {
public NettyServerHandler(ThreadPoolTaskExecutor threadPoolExecutor) {
this.threadPoolExecutor = threadPoolExecutor;
}
@Override
public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
ctx.channel().writeAndFlush("2a320d0a24340d0a415554480d0a2431360d0a536d2a674234504742333939514241390d0a");
ctx.channel().writeAndFlush("2a320d0a24340d0a415554480d0a2431360d0a536d2a674234504742333939514241390d0a2a330d0a24360d0a434c49454e540d0a24370d0a5345544e414d450d0a2431310d0a574a2d31303530333036320d0a2a320d0a24340d0a4543484f0d0a2431360d0a5135ee72dcef1143abbebbd4de9e5cbf0d0a2a320d0a24390d0a5355425343524942450d0a2432360d0a5f5f426f6f6b736c656576655f4d61737465724368616e6765640d0a");
ctx.fireChannelActive();
}
/**
* 建立连接后
*
* @param ctx
* @throws Exception
*/
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
log.info("{} 连接成功....", ctx.channel().remoteAddress());
ctx.channel().writeAndFlush("2a320d0a24340d0a415554480d0a2431360d0a536d2a674234504742333939514241390d0a");
ctx.channel().writeAndFlush("2a320d0a24340d0a415554480d0a2431360d0a536d2a674234504742333939514241390d0a2a330d0a24360d0a434c49454e540d0a24370d0a5345544e414d450d0a2431310d0a574a2d31303530333036320d0a2a320d0a24340d0a4543484f0d0a2431360d0a5135ee72dcef1143abbebbd4de9e5cbf0d0a2a320d0a24390d0a5355425343524942450d0a2432360d0a5f5f426f6f6b736c656576655f4d61737465724368616e6765640d0a");
ctx.fireChannelActive();
}
@Autowired
CrossInfoCache crossInfoCache;
@Autowired
ThreadPoolTaskExecutor threadPoolExecutor;
/**
* 读取消息
*
* @param ctx
* @param message
* @param packet
* @throws Exception
*/
@Override
public void channelRead(ChannelHandlerContext ctx, Object message) throws Exception {
String key = ctx.channel().remoteAddress().toString();
String data = message.toString();
log.info("收到数据体: {}", data);
log.info("收到消息地址： {}", key);
String hexSign = AgreementCommon.getHexSign(data);
protected void channelRead0(ChannelHandlerContext ctx, DatagramPacket packet) throws Exception {
log.info("收到服务端包数据： {}", packet);
ByteBuf content = packet.content();
InetSocketAddress remote = packet.sender();
int i = content.readableBytes();
byte[] b = new byte[i];
content.readBytes(b);
String data = HexUtil.encodeHexStr(b);
String hexSign = CommandResultSign.getHexSign(data);
if (StringUtils.isEmpty(hexSign)) {
log.error("data unique hex sign error");
return;
}
String className = CommandResultSign.getClassNameByHexSign(hexSign);
CommandResponseService commandResponseService = (CommandResponseService) ServiceBeanContext.getBean("heartBeatService");
String className = getClassNameByHexSign(hexSign);
CommandResponseService commandResponseService = (CommandResponseService) ServiceBeanContext.getBean(className);
String key = StringUtils.join("/", remote.getHostString(), ":", remote.getPort(), "/", hexSign);
CommandPojo commandPojo = getCommandPojo(key, data);
String hexResult = commandResponseService.getCommandResponse(commandPojo);
if (Objects.nonNull(hexResult)) {
......@@ -88,12 +73,16 @@ public class NettyServerHandler extends ChannelInboundHandlerAdapter {
}
private CommandPojo getCommandPojo(String key, String data) {
CrossInfoPO crossInfo = crossInfoCache.getCrossInfo(key);
// CrossInfoPO crossInfo = crossInfoCache.getCrossInfo(key);
CommandPojo commandPojo = new CommandPojo();
commandPojo.setCrossId(crossInfo.getId());
commandPojo.setSignalId(crossInfo.getCode());
commandPojo.setSignalIp(crossInfo.getIp());
commandPojo.setPort(crossInfo.getPort());
// commandPojo.setCrossId(crossInfo.getId());
// commandPojo.setSignalId(crossInfo.getCode());
// commandPojo.setSignalIp(crossInfo.getIp());
// commandPojo.setPort(crossInfo.getPort());
commandPojo.setCrossId("1111");
commandPojo.setSignalId("2222");
commandPojo.setSignalIp("10.102.1.204");
commandPojo.setPort(5050);
commandPojo.setResponseMsg(data);
return commandPojo;
}
......
signal-utc-dt-service/src/main/java/net/wanji/com/netty/response/impl/HeartBeatService.java
View file @ 7439d1d4
package net.wanji.com.netty.response.impl;
import cn.hutool.core.util.HexUtil;
import lombok.extern.slf4j.Slf4j;
import net.wanji.com.common.constants.Constants;
import net.wanji.com.common.enums.CommandResultSign;
import net.wanji.com.netty.NettyClient;
import net.wanji.com.netty.pojo.CommandPojo;
import net.wanji.com.netty.response.CommandResponseService;
import net.wanji.com.pojo.netty.MessageResultPojo;
import net.wanji.com.util.CRC16Utils;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
/**
......@@ -20,43 +17,18 @@ import org.springframework.stereotype.Service;
public class HeartBeatService implements CommandResponseService {
@Override
public String getCommandResponse(CommandPojo commandPojo) {
String body = String.format("%02x", Constants.COMMAND_HEARTBEAT.length() / 2).concat(Constants.COMMAND_HEARTBEAT);
String crc16HexStr = CRC16Utils.getCRC16HexStr(body);
String heartBeatRequest = Constants.buildMessageBody(body, crc16HexStr);
NettyClient.sendMessage(commandPojo.getSignalIp(), commandPojo.getPort(), heartBeatRequest, CommandResultSign.HEARTBEAT.getHexSign(), 300);
return heartBeatRequest;
return commandPojo.getResponseMsg();
}
// @Scheduled(fixedRate = 1000)
@Scheduled(fixedRate = 1000)
public void heartbeat() {
String body = String.format("%02x", Constants.COMMAND_HEARTBEAT.length() / 2).concat(Constants.COMMAND_HEARTBEAT);
String crc16HexStr = CRC16Utils.getCRC16HexStr(body);
String heartBeatRequest = Constants.buildMessageBody(body, crc16HexStr);
// String str = "55ac00000000000017050f11060b0000000000000000000000000000000000";
String str = "2a320d0a24340d0a494e464f0d0a2431310d0a7265706c69636174696f6e0d0a";
MessageResultPojo resultPojo = NettyClient.sendMessage("10.102.1.204", 5050, str, "80", 300);
log.info("信号返回字符串：{}", resultPojo);
String str1 = "2a310d0a24340d0a50494e470d0a";
MessageResultPojo resultPojo1 = NettyClient.sendMessage("10.102.1.204", 5050, str1, "80", 300);
log.info("信号返回字符串1：{}", resultPojo1);
// String body = String.format("%02x", Constants.COMMAND_HEARTBEAT.length() / 2).concat(Constants.COMMAND_HEARTBEAT);
// String crc16HexStr = CRC16Utils.getCRC16HexStr(body);
// String heartBeatRequest = Constants.buildMessageBody(body, crc16HexStr);
// MessageResultPojo resultPojo2 = NettyClient.sendMessage("10.102.1.204", 5050, heartBeatRequest, "55ab", 300);
String str = "55ac00000000000017050f11060b0000000000000000000000000000000000";
MessageResultPojo resultPojo = NettyClient.sendMessage("10.102.1.204", 5050, str, "55ab", 300);
log.info("信号返回字符串：{}", resultPojo.getHexMessageResult());
}
public static void main(String[] args) {
String body = String.format("%04x", Constants.COMMAND_HEARTBEAT.length() / 2).concat(Constants.COMMAND_HEARTBEAT);
String crc16HexStr = CRC16Utils.getCRC16HexStr(body);
String heartBeatRequest = Constants.buildMessageBody(body, crc16HexStr);
System.err.println(heartBeatRequest);
// NettyServer.sendMessage(commandPojo.getSignalIp(), commandPojo.getPort(), heartBeatRequest, CommandResultSign.HEARTBEAT.getHexSign(), 300);
String str = "55AB00A7000000000000000002000000FFFFFFFF00000000FFFFFFFF01000000FFFFFFFF1010101004101010101010100410101000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000";
byte[] bytes = str.getBytes();
System.err.println(HexUtil.encodeHexStr(bytes));
for (byte aByte : bytes) {
System.err.println(HexUtil.toHex(aByte));
}
}
}
signal-utc-dt-service/src/main/java/net/wanji/com/util/CRC16Utils.java
View file @ 7439d1d4
......@@ -10,7 +10,7 @@ import cn.hutool.core.util.HexUtil;
public class CRC16Utils {
/**
* 获取CRC16校验结果
* 获取CRC16校验结果 【报文长度+报文体】
*
* @param body
* @return
......
signal-utc-dt-service/src/main/resources/application-dev.yml
View file @ 7439d1d4
......@@ -10,13 +10,6 @@ server:
threads:
max: 10000
#线程池配置
threadPool:
corePoolSize: 5
maxPoolSize: 200
queueCapacity: 5
keepAliveTime: 60000
spring:
datasource:
dynamic:
......
signal-utc-dt-service/src/main/resources/application.yml
View file @ 7439d1d4
......@@ -61,7 +61,7 @@ mybatis-plus:
#线程池配置
threadPoolConfig:
threadPoolName: threadPoolExecutor
threadPoolName: ${spring.application.name}-threadPool
coreSize: 8
maxSize: 16
queueCapacity: 200
......
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