Browse Source

fix

master
winter 2 hours ago
parent
commit
0001eb6253
13 changed files with 473 additions and 14 deletions
  1. +1
    -0
      bjairCApi/pom.xml
  2. +14
    -1
      bjairCApi/src/main/java/com/iformall/CApplication.java
  3. +1
    -1
      bjairCApi/src/main/java/com/iformall/controller/haikang/TcpClient.java
  4. +28
    -12
      bjairCApi/src/main/java/com/iformall/controller/haikang/TcpServer.java
  5. +7
    -0
      bjairService/pom.xml
  6. +17
    -0
      bjairService/src/main/java/com/iformall/haikang/netty/HexMessageDecoder.java
  7. +15
    -0
      bjairService/src/main/java/com/iformall/haikang/netty/HexMessageEncoder.java
  8. +93
    -0
      bjairService/src/main/java/com/iformall/haikang/netty/HexUtils.java
  9. +41
    -0
      bjairService/src/main/java/com/iformall/haikang/netty/IpRejectHandler.java
  10. +132
    -0
      bjairService/src/main/java/com/iformall/haikang/netty/NioSocketHandler.java
  11. +55
    -0
      bjairService/src/main/java/com/iformall/haikang/netty/NioSocketInitializer.java
  12. +48
    -0
      bjairService/src/main/java/com/iformall/haikang/netty/NioSocketServer.java
  13. +21
    -0
      bjairService/src/main/java/com/iformall/haikang/netty/ThreadPoolConfig.java

+ 1
- 0
bjairCApi/pom.xml View File

@@ -45,6 +45,7 @@
<artifactId>flyway-core</artifactId>
<version>5.2.4</version>
</dependency>
</dependencies>
<build>
<plugins>


+ 14
- 1
bjairCApi/src/main/java/com/iformall/CApplication.java View File

@@ -1,9 +1,13 @@
package com.iformall;

import com.iformall.annotation.BaseEnableSwagger;
import com.iformall.haikang.netty.NioSocketServer;

import org.mybatis.spring.annotation.MapperScan;
import org.rocketmq.starter.annotation.EnableRocketMQ;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
@@ -23,7 +27,7 @@ import org.springframework.scheduling.annotation.EnableAsync;
@EnableAsync
@EnableRocketMQ
@EnableAspectJAutoProxy(exposeProxy = true)
public class CApplication {
public class CApplication implements CommandLineRunner{

@Value("${fm.exception}")
private boolean fmException;
@@ -45,6 +49,15 @@ public class CApplication {
@Value("${tcp.server.port}")
private int tcpServerPort;
@Autowired
private NioSocketServer nioSocketServer;
@Override
public void run(String... args) throws Exception {
nioSocketServer.start(tcpServerPort);
}

@Bean
public boolean isFmException() {


+ 1
- 1
bjairCApi/src/main/java/com/iformall/controller/haikang/TcpClient.java View File

@@ -13,7 +13,7 @@ import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

@Component
//@Component
public class TcpClient {
private final Logger log = LoggerFactory.getLogger(this.getClass());


+ 28
- 12
bjairCApi/src/main/java/com/iformall/controller/haikang/TcpServer.java View File

@@ -23,7 +23,7 @@ import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.iformall.haikang.HKPeopleNumberFactory;

@Component
//@Component
public class TcpServer {
private final Logger log = LoggerFactory.getLogger(this.getClass());
@@ -54,26 +54,42 @@ public class TcpServer {
}

private void handleClient(Socket clientSocket) {
try (
BufferedReader in = new BufferedReader(new InputStreamReader(clientSocket.getInputStream()));
PrintWriter out = new PrintWriter(clientSocket.getOutputStream(), true)
) {
BufferedReader in = null;
PrintWriter out = null;
try {
in = new BufferedReader(new InputStreamReader(clientSocket.getInputStream()));
out = new PrintWriter(clientSocket.getOutputStream(), true);
String message;
while ((message = in.readLine()) != null) {
log.info("Received: {}", message);
//
handleMessage(out, message);
//handleMessage(out, message);
out.flush();
// Echo back to client
//out.println("Server: " + message);
}
} catch (IOException e) {
log.error("Client connection error: {}", e.getMessage());
log.error("Client connection error: {}", e);
} finally {
try {
clientSocket.close();
} catch (IOException e) {
log.error("Failed to close socket: {}", e.getMessage());
}
// try {
// if (null != in) {
// in.close();
// }
// } catch (Exception e) {
// log.error("Failed to close socket: {}", e);
// }
// try {
// if (null != out) {
// out.close();
// }
// } catch (Exception e) {
// log.error("Failed to close socket: {}", e);
// }
// try {
// clientSocket.close();
// } catch (IOException e) {
// log.error("Failed to close socket: {}", e);
// }
}
}


+ 7
- 0
bjairService/pom.xml View File

@@ -52,6 +52,13 @@
<artifactId>jsoup</artifactId>
<version>1.10.2</version>
</dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.1.68.Final</version>
<!-- 使用适合你项目的版本 -->
</dependency>
</dependencies>

</project>

+ 17
- 0
bjairService/src/main/java/com/iformall/haikang/netty/HexMessageDecoder.java View File

@@ -0,0 +1,17 @@
package com.iformall.haikang.netty;

import java.util.List;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToMessageDecoder;

public class HexMessageDecoder extends MessageToMessageDecoder<ByteBuf>{
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
byte[] bytes = new byte[in.readableBytes()];
in.readBytes(bytes);
String hexString = HexUtils.byteArrayToHexString(bytes);
out.add(hexString);
}
}

+ 15
- 0
bjairService/src/main/java/com/iformall/haikang/netty/HexMessageEncoder.java View File

@@ -0,0 +1,15 @@
package com.iformall.haikang.netty;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;

public class HexMessageEncoder extends MessageToByteEncoder<String>{
@Override
protected void encode(ChannelHandlerContext ctx, String msg, ByteBuf out) throws Exception {
byte[] bytes = HexUtils.hexStringToByteArray(msg);
ByteBuf buffer = Unpooled.wrappedBuffer(bytes);
out.writeBytes(buffer);
}
}

+ 93
- 0
bjairService/src/main/java/com/iformall/haikang/netty/HexUtils.java View File

@@ -0,0 +1,93 @@
package com.iformall.haikang.netty;

public class HexUtils {
public static String byteArrayToHexString(byte[] data) {
StringBuilder hexString = new StringBuilder();
for (byte b : data) {
hexString.append(String.format("%02X ", b));
}
return hexString.toString().trim();
}

public static byte[] hexStringToByteArray(String s) {
s = s.replaceAll("\\s", ""); // 去除所有空白字符
int len = s.length();
if (len % 2 != 0) {
throw new IllegalArgumentException("Invalid hexadecimal string length");
}

byte[] data = new byte[len / 2];

for (int i = 0; i < len; i += 2) {
data[i / 2] = (byte) ((Character.digit(s.charAt(i), 16) << 4)
+ Character.digit(s.charAt(i + 1), 16));
}
return data;
}

/**
* 将16进制字符串转换为ASCII字符串
* @param hexString
* @return
*/
public static String hexStringToAscii(String hexString) {
byte[] bytes = hexStringToByteArray(hexString);
return new String(bytes);
}

/**
* 将16进制字符串转换为十进制整型
* @param hexString
* @return
*/
public static long hexStringToDecimal(String hexString) {
hexString = hexString.replaceAll("\\s", ""); // 去除所有空白字符
long decimalValue = 0;

for (int i = 0; i < hexString.length(); i++) {
char c = hexString.charAt(i);
int value = Character.digit(c, 16); // 获取当前字符的10进制值
decimalValue = decimalValue * 16 + value;
}

return decimalValue;
}


/**
* 将16进制字符串转换为浮点数
* @param hexString
* @return
*/
public static float hexStringToFloat(String hexString) {
hexString = hexString.replaceAll("\\s", ""); // 去除所有空白字符
byte[] bytes = hexStringToByteArray(hexString);
return bytesToFloat(bytes);
}

/**
* 格式化16进制字符串
* @param hexString
* @return
*/
public static String formatHexString(String hexString) {
StringBuilder formattedString = new StringBuilder();
for (int i = 0; i < hexString.length(); i += 2) {
if (i > 0) {
formattedString.append(" ");
}
formattedString.append(hexString.substring(i, i + 2));
}

return formattedString.toString();
}

private static float bytesToFloat(byte[] bytes) {
int intBits = 0;
for (int i = 0; i < 4; i++) {
intBits |= (bytes[i] & 0xFF) << (24 - 8 * i);
}
return Float.intBitsToFloat(intBits);
}

}

+ 41
- 0
bjairService/src/main/java/com/iformall/haikang/netty/IpRejectHandler.java View File

@@ -0,0 +1,41 @@
package com.iformall.haikang.netty;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

public class IpRejectHandler extends ChannelInboundHandlerAdapter{
private final Logger log = LoggerFactory.getLogger(this.getClass());
// 拒绝连接的IP列表
private static final String[] BLOCKED_IPS = {};

@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
String remoteAddress = ctx.channel().remoteAddress().toString();
if (isBlocked(remoteAddress)) {
log.error("netty tcp 拒绝连接: " + remoteAddress);
ctx.close();
} else {
log.info("netty tcp 允许连接: " + remoteAddress);
}
}

@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
String remoteAddress = ctx.channel().remoteAddress().toString();
log.info("netty tcp 设备已断开连接: " + remoteAddress);
super.channelInactive(ctx);
}

private boolean isBlocked(String remoteAddress) {
for (String blockedIp : BLOCKED_IPS) {
if (remoteAddress.contains(blockedIp)) {
return true;
}
}
return false;
}
}

+ 132
- 0
bjairService/src/main/java/com/iformall/haikang/netty/NioSocketHandler.java View File

@@ -0,0 +1,132 @@
package com.iformall.haikang.netty;

import java.io.PrintWriter;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;

import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.iformall.haikang.HKPeopleNumberFactory;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

@Component
public class NioSocketHandler extends SimpleChannelInboundHandler<String>{
private final Logger log = LoggerFactory.getLogger(this.getClass());

private String hexData = "";
private HKPeopleNumberFactory peopleNumberFactory;
private final ThreadPoolTaskExecutor threadPoolTaskExecutor;

public NioSocketHandler(ThreadPoolTaskExecutor threadPoolTaskExecutor,HKPeopleNumberFactory peopleNumberFactory) {
this.threadPoolTaskExecutor = threadPoolTaskExecutor;
this.peopleNumberFactory = peopleNumberFactory;
}

@Override
protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
// 异步处理消息
threadPoolTaskExecutor.execute(() -> {
// 你的处理逻辑
String clientId = ctx.channel().id().asShortText();
log.info("netty tcp 接收到用户id为"+clientId+"的消息: " + msg);

// 应答报文,根据要求应答
String repeak = msg;
handleMessage(ctx,msg);
// 应答
//ctx.writeAndFlush(repeak);

});
}

@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
//编号2代表需要等待时长,编号3代表当前排队人数,编号4代表当前场内出租车,编号5代表在途出租车
private static final int[] types = new int[] {2,3,4,5};
private void handleMessage(ChannelHandlerContext ctx,String message) {
if (StringUtils.isBlank(message)) {
return ;
}
JSONObject mo = JSON.parseObject(message);
if (null == mo) {
return ;
}
String func = mo.getString("func");
//心跳
if ("KeepOnline".equals(func)) {
Map keepMap = new LinkedHashMap();
String dest = mo.getString("dest");
String source = mo.getString("source");
keepMap.put("dest", source);
keepMap.put("source", dest);
keepMap.put("func", "AckKeepOnline");
JSONArray datas = new JSONArray();
Map dm = new HashMap();
dm.put("node", "OK");
datas.add(dm);
keepMap.put("data", datas);
String xintiaoresult = JSON.toJSONString(keepMap);
log.debug("haikang keepOnline result:"+xintiaoresult);
ctx.writeAndFlush(xintiaoresult);
//给灯箱发送
for (int i = 0 ; i < types.length ; i++) {
Map dxobj = new LinkedHashMap();
dxobj.put("dest", source);
dxobj.put("source", dest);
dxobj.put("func", "ControlData");
JSONArray datas1 = new JSONArray();
Map dm1 = new HashMap();
dm1.put("node", generateNode(types[i],source));
datas1.add(dm1);
dxobj.put("data", datas1);
String dengxiangcontent = JSON.toJSONString(dxobj);
log.debug("haikang send result:"+dengxiangcontent);
ctx.writeAndFlush(dengxiangcontent);
}
}
}
//编号2代表需要等待时长,编号3代表当前排队人数,编号4代表当前场内出租车,编号5代表在途出租车
private String generateNode(int type,String sn) {
StringBuffer sb = new StringBuffer("0-5-").append(type).append("-");
int value = 0;
switch (type) {
case 2:
value = 1;//查询等待时长
break;
case 3:
value = peopleNumberFactory.getPeopleNumber();;//查询当前排队人数
break;
case 4:
value = 1;//查询场内出租车
break;
case 5:
value = 1;//查询在途出租车
break;
default:
break;
}
sb.append(value).append("-0-1-"+sn);
return sb.toString();
}
}

+ 55
- 0
bjairService/src/main/java/com/iformall/haikang/netty/NioSocketInitializer.java View File

@@ -0,0 +1,55 @@
package com.iformall.haikang.netty;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import com.iformall.haikang.HKPeopleNumberFactory;

import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;

@Component
public class NioSocketInitializer extends ChannelInitializer<SocketChannel>{
private final Logger log = LoggerFactory.getLogger(this.getClass());

@Autowired
private ThreadPoolTaskExecutor threadPoolTaskExecutor;
@Autowired
private HKPeopleNumberFactory peopleNumberFactory;

public NioSocketInitializer(ThreadPoolTaskExecutor threadPoolTaskExecutor) {
this.threadPoolTaskExecutor = threadPoolTaskExecutor;
}


@Override
protected void initChannel(SocketChannel ch) {
ChannelPipeline pipeline = ch.pipeline();
// tcp报文解码器,接收后自动调用解码
//pipeline.addLast(new HexMessageDecoder());
// tcp报文编码器,应答时自动编码
//pipeline.addLast(new HexMessageEncoder());

// 如果socket传输的是字符串类型,请使用StringDecoder和StringEncoder
pipeline.addLast(new StringDecoder());
pipeline.addLast(new StringEncoder());

// 初始化并添加你的业务处理类
//ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
//threadPoolTaskExecutor.initialize(); // 初始化线程池

// 使用自定义线程池处理业务逻辑
log.info(">>>>peopleNumberFactory:"+peopleNumberFactory);
pipeline.addLast(new NioSocketHandler(threadPoolTaskExecutor,peopleNumberFactory));
// 连接策略
pipeline.addLast(new IpRejectHandler());
}

}

+ 48
- 0
bjairService/src/main/java/com/iformall/haikang/netty/NioSocketServer.java View File

@@ -0,0 +1,48 @@
package com.iformall.haikang.netty;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;

@Component
public class NioSocketServer {
private final Logger log = LoggerFactory.getLogger(this.getClass());
@Autowired
private ThreadPoolTaskExecutor threadPoolTaskExecutor;

private EventLoopGroup bossGroup;
private EventLoopGroup workerGroup;
private ChannelFuture channelFuture;

public void start(int port) throws Exception {
bossGroup = new NioEventLoopGroup(1); // 主线程组
workerGroup = new NioEventLoopGroup(); // 工作线程组

try {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new NioSocketInitializer(threadPoolTaskExecutor))
.option(ChannelOption.SO_BACKLOG, 128)
.childOption(ChannelOption.SO_KEEPALIVE, true);

channelFuture = b.bind(port).sync();
log.info("netty tcp NIO Socket服务器已启动,监听端口:" + port);

channelFuture.channel().closeFuture().sync();
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
}

+ 21
- 0
bjairService/src/main/java/com/iformall/haikang/netty/ThreadPoolConfig.java View File

@@ -0,0 +1,21 @@
package com.iformall.haikang.netty;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

@Configuration
public class ThreadPoolConfig {

@Bean
public ThreadPoolTaskExecutor threadPoolTaskExecutor() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(10); // 核心线程数
executor.setMaxPoolSize(20); // 最大线程数
executor.setQueueCapacity(100); // 队列容量
executor.setThreadNamePrefix("NIOSocket-ThreadPool-"); // 线程前缀
executor.initialize();
return executor;
}

}

Loading…
Cancel
Save