33 changed files with 2541 additions and 2571 deletions
@ -1,41 +1,41 @@ |
|||||||
package com.mh.user.job; |
//package com.mh.user.job;
|
||||||
|
//
|
||||||
import com.mh.user.netty.NettyClient; |
//import com.mh.user.netty.NettyClient;
|
||||||
import com.mh.user.constants.SocketMessage; |
//import com.mh.user.constants.SocketMessage;
|
||||||
import lombok.SneakyThrows; |
//import lombok.SneakyThrows;
|
||||||
import lombok.extern.slf4j.Slf4j; |
//import lombok.extern.slf4j.Slf4j;
|
||||||
import org.quartz.DisallowConcurrentExecution; |
//import org.quartz.DisallowConcurrentExecution;
|
||||||
import org.quartz.Job; |
//import org.quartz.Job;
|
||||||
import org.quartz.JobExecutionContext; |
//import org.quartz.JobExecutionContext;
|
||||||
import org.quartz.JobExecutionException; |
//import org.quartz.JobExecutionException;
|
||||||
import org.springframework.beans.factory.annotation.Autowired; |
//import org.springframework.beans.factory.annotation.Autowired;
|
||||||
|
//
|
||||||
/** |
///**
|
||||||
* @author ljf |
// * @author ljf
|
||||||
* @title : |
// * @title :
|
||||||
* @description : 定时采集冷量计任务 |
// * @description : 定时采集冷量计任务
|
||||||
* @updateTime 2020-05-18 |
// * @updateTime 2020-05-18
|
||||||
* @throws : |
// * @throws :
|
||||||
*/ |
// */
|
||||||
/** |
///**
|
||||||
* :@DisallowConcurrentExecution : 此标记用在实现Job的类上面,意思是不允许并发执行. |
// * :@DisallowConcurrentExecution : 此标记用在实现Job的类上面,意思是不允许并发执行.
|
||||||
* :注意org.quartz.threadPool.threadCount线程池中线程的数量至少要多个,否则@DisallowConcurrentExecution不生效 |
// * :注意org.quartz.threadPool.threadCount线程池中线程的数量至少要多个,否则@DisallowConcurrentExecution不生效
|
||||||
* :假如Job的设置时间间隔为3秒,但Job执行时间是5秒,设置@DisallowConcurrentExecution以后程序会等任务执行完毕以后再去执行, |
// * :假如Job的设置时间间隔为3秒,但Job执行时间是5秒,设置@DisallowConcurrentExecution以后程序会等任务执行完毕以后再去执行,
|
||||||
* 否则会在3秒时再启用新的线程执行 |
// * 否则会在3秒时再启用新的线程执行
|
||||||
*/ |
// */
|
||||||
@DisallowConcurrentExecution |
//@DisallowConcurrentExecution
|
||||||
@Slf4j |
//@Slf4j
|
||||||
public class JobCloud implements Job { |
//public class JobCloud implements Job {
|
||||||
|
//
|
||||||
@Autowired |
// @Autowired
|
||||||
private SocketMessage socketMessage; |
// private SocketMessage socketMessage;
|
||||||
|
//
|
||||||
@SneakyThrows |
// @SneakyThrows
|
||||||
@Override |
// @Override
|
||||||
public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException { |
// public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException {
|
||||||
// 定时采集冷量计
|
// // 定时采集冷量计
|
||||||
log.info("定时采集冷量计"); |
// log.info("定时采集冷量计");
|
||||||
NettyClient nettyClient = new NettyClient(); |
// NettyClient nettyClient = new NettyClient();
|
||||||
nettyClient.connect(socketMessage.getPort(),socketMessage.getIP()); |
// nettyClient.connect(socketMessage.getPort(),socketMessage.getIP());
|
||||||
} |
// }
|
||||||
} |
//}
|
||||||
|
@ -1,39 +1,39 @@ |
|||||||
package com.mh.user.job; |
//package com.mh.user.job;
|
||||||
|
//
|
||||||
import com.mh.user.netty.NettyMeterClient; |
//import com.mh.user.netty.NettyMeterClient;
|
||||||
import com.mh.user.constants.SocketMessage; |
//import com.mh.user.constants.SocketMessage;
|
||||||
import lombok.SneakyThrows; |
//import lombok.SneakyThrows;
|
||||||
import lombok.extern.slf4j.Slf4j; |
//import lombok.extern.slf4j.Slf4j;
|
||||||
import org.quartz.DisallowConcurrentExecution; |
//import org.quartz.DisallowConcurrentExecution;
|
||||||
import org.quartz.Job; |
//import org.quartz.Job;
|
||||||
import org.quartz.JobExecutionContext; |
//import org.quartz.JobExecutionContext;
|
||||||
import org.springframework.beans.factory.annotation.Autowired; |
//import org.springframework.beans.factory.annotation.Autowired;
|
||||||
|
//
|
||||||
/** |
///**
|
||||||
* @author ljf |
// * @author ljf
|
||||||
* @title : |
// * @title :
|
||||||
* @description : 定时采集电表数据任务 |
// * @description : 定时采集电表数据任务
|
||||||
* @updateTime 2020-05-18 |
// * @updateTime 2020-05-18
|
||||||
* @throws : |
// * @throws :
|
||||||
*/ |
// */
|
||||||
/** |
///**
|
||||||
* :@DisallowConcurrentExecution : 此标记用在实现Job的类上面,意思是不允许并发执行. |
// * :@DisallowConcurrentExecution : 此标记用在实现Job的类上面,意思是不允许并发执行.
|
||||||
* :注意org.quartz.threadPool.threadCount线程池中线程的数量至少要多个,否则@DisallowConcurrentExecution不生效 |
// * :注意org.quartz.threadPool.threadCount线程池中线程的数量至少要多个,否则@DisallowConcurrentExecution不生效
|
||||||
* :假如Job的设置时间间隔为3秒,但Job执行时间是5秒,设置@DisallowConcurrentExecution以后程序会等任务执行完毕以后再去执行, |
// * :假如Job的设置时间间隔为3秒,但Job执行时间是5秒,设置@DisallowConcurrentExecution以后程序会等任务执行完毕以后再去执行,
|
||||||
* 否则会在3秒时再启用新的线程执行 |
// * 否则会在3秒时再启用新的线程执行
|
||||||
*/ |
// */
|
||||||
@DisallowConcurrentExecution |
//@DisallowConcurrentExecution
|
||||||
@Slf4j |
//@Slf4j
|
||||||
public class JobMeter implements Job { |
//public class JobMeter implements Job {
|
||||||
|
//
|
||||||
@Autowired |
// @Autowired
|
||||||
private SocketMessage socketMessage; |
// private SocketMessage socketMessage;
|
||||||
|
//
|
||||||
@SneakyThrows |
// @SneakyThrows
|
||||||
@Override |
// @Override
|
||||||
public void execute(JobExecutionContext jobExecutionContext) { |
// public void execute(JobExecutionContext jobExecutionContext) {
|
||||||
log.info("定时采集电表数据任务开始"); |
// log.info("定时采集电表数据任务开始");
|
||||||
NettyMeterClient nettyMeterClient = new NettyMeterClient(); |
// NettyMeterClient nettyMeterClient = new NettyMeterClient();
|
||||||
nettyMeterClient.connect(socketMessage.getPort(), socketMessage.getIP()); |
// nettyMeterClient.connect(socketMessage.getPort(), socketMessage.getIP());
|
||||||
} |
// }
|
||||||
} |
//}
|
||||||
|
@ -1,295 +1,295 @@ |
|||||||
package com.mh.user.netty; |
//package com.mh.user.netty;
|
||||||
|
//
|
||||||
import com.mh.user.constants.Constant; |
//import com.mh.user.constants.Constant;
|
||||||
import com.mh.user.entity.DeviceManageEntity; |
//import com.mh.user.entity.DeviceManageEntity;
|
||||||
import com.mh.user.service.DeviceManageService; |
//import com.mh.user.service.DeviceManageService;
|
||||||
import com.mh.user.utils.*; |
//import com.mh.user.utils.*;
|
||||||
import io.netty.buffer.ByteBuf; |
//import io.netty.buffer.ByteBuf;
|
||||||
import io.netty.channel.Channel; |
//import io.netty.channel.Channel;
|
||||||
import io.netty.channel.ChannelHandlerAdapter; |
//import io.netty.channel.ChannelHandlerAdapter;
|
||||||
import io.netty.channel.ChannelHandlerContext; |
//import io.netty.channel.ChannelHandlerContext;
|
||||||
import io.netty.util.ReferenceCountUtil; |
//import io.netty.util.ReferenceCountUtil;
|
||||||
import lombok.extern.slf4j.Slf4j; |
//import lombok.extern.slf4j.Slf4j;
|
||||||
import org.springframework.context.ApplicationContext; |
//import org.springframework.context.ApplicationContext;
|
||||||
|
//
|
||||||
import java.text.SimpleDateFormat; |
//import java.text.SimpleDateFormat;
|
||||||
import java.util.Date; |
//import java.util.Date;
|
||||||
import java.util.List; |
//import java.util.List;
|
||||||
|
//
|
||||||
/** |
///**
|
||||||
* @author ljf |
// * @author ljf
|
||||||
* @title : |
// * @title :
|
||||||
* @description :客户端异步消息处理机制 |
// * @description :客户端异步消息处理机制
|
||||||
* @updateTime 2020-05-13 |
// * @updateTime 2020-05-13
|
||||||
* @throws : |
// * @throws :
|
||||||
*/ |
// */
|
||||||
@Slf4j |
//@Slf4j
|
||||||
public class NettyClientHandler extends ChannelHandlerAdapter { |
//public class NettyClientHandler extends ChannelHandlerAdapter {
|
||||||
|
//
|
||||||
private int num = 0; |
// private int num = 0;
|
||||||
private int size = 0; |
// private int size = 0;
|
||||||
private String receiveStr = null; |
// private String receiveStr = null;
|
||||||
private String IP = null; |
// private String IP = null;
|
||||||
private String port = null; |
// private String port = null;
|
||||||
List<DeviceManageEntity> deviceManageEntityList; |
// List<DeviceManageEntity> deviceManageEntityList;
|
||||||
|
//
|
||||||
// 调用service
|
// // 调用service
|
||||||
ApplicationContext context = SpringBeanUtil.getApplicationContext(); |
// ApplicationContext context = SpringBeanUtil.getApplicationContext();
|
||||||
DeviceManageService deviceManageService = context.getBean(DeviceManageService.class); |
// DeviceManageService deviceManageService = context.getBean(DeviceManageService.class);
|
||||||
|
//
|
||||||
|
//
|
||||||
@Override |
// @Override
|
||||||
public void channelUnregistered(ChannelHandlerContext ctx) throws Exception { |
// public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
|
||||||
log.info("当前channel从EventLoop取消注册"); |
// log.info("当前channel从EventLoop取消注册");
|
||||||
super.channelUnregistered(ctx); |
// super.channelUnregistered(ctx);
|
||||||
} |
// }
|
||||||
|
//
|
||||||
@Override |
// @Override
|
||||||
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { |
// public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
|
||||||
// super.exceptionCaught(ctx, cause);
|
//// super.exceptionCaught(ctx, cause);
|
||||||
log.info("通信异常!!"); |
// log.info("通信异常!!");
|
||||||
receiveStr = null; |
|
||||||
Channel incoming = ctx.channel(); |
|
||||||
if (incoming.isActive()){ |
|
||||||
log.info("SimpleClient: " + incoming.remoteAddress() + "异常"); |
|
||||||
cause.printStackTrace(); |
|
||||||
ctx.close(); |
|
||||||
// receiveStr = null;
|
// receiveStr = null;
|
||||||
// try {
|
// Channel incoming = ctx.channel();
|
||||||
// TimeUnit.SECONDS.sleep(5);
|
// if (incoming.isActive()){
|
||||||
// SocketAddress remoteAddress = ctx.channel().remoteAddress();
|
// log.info("SimpleClient: " + incoming.remoteAddress() + "异常");
|
||||||
// String port = ExchangeStringUtil.endData(remoteAddress.toString(),":");
|
// cause.printStackTrace();
|
||||||
// String host = ExchangeStringUtil.splitData(remoteAddress.toString(),"/",":");
|
// ctx.close();
|
||||||
// NettyClient nettyClient = new NettyClient();
|
//// receiveStr = null;
|
||||||
// nettyClient.connect(Integer.parseInt(port), host); // 断线重连
|
//// try {
|
||||||
// } catch (InterruptedException e) {
|
//// TimeUnit.SECONDS.sleep(5);
|
||||||
// e.printStackTrace();
|
//// SocketAddress remoteAddress = ctx.channel().remoteAddress();
|
||||||
|
//// String port = ExchangeStringUtil.endData(remoteAddress.toString(),":");
|
||||||
|
//// String host = ExchangeStringUtil.splitData(remoteAddress.toString(),"/",":");
|
||||||
|
//// NettyClient nettyClient = new NettyClient();
|
||||||
|
//// nettyClient.connect(Integer.parseInt(port), host); // 断线重连
|
||||||
|
//// } catch (InterruptedException e) {
|
||||||
|
//// e.printStackTrace();
|
||||||
|
//// }
|
||||||
// }
|
// }
|
||||||
} |
// }
|
||||||
} |
//
|
||||||
|
|
||||||
|
|
||||||
@Override |
|
||||||
public void channelActive(ChannelHandlerContext ctx) { |
|
||||||
// 添加一个状态值,判断是否继续发送指令 update by ljf on 2020-08-07
|
|
||||||
if (Constant.WEB_FLAG) { |
|
||||||
num = 0; |
|
||||||
// 关闭连接
|
|
||||||
receiveStr = null; |
|
||||||
ctx.close(); |
|
||||||
} else { |
|
||||||
ctx.channel().read(); |
|
||||||
SimpleDateFormat sdf1 = new SimpleDateFormat("yyyy-mm-dd HH:mm:ss"); |
|
||||||
Date date = new Date(); |
|
||||||
log.info(ctx.channel().remoteAddress() + " " + sdf1.format(date) + "链接服务端成功!"); |
|
||||||
// 截取IP地址
|
|
||||||
IP = ExchangeStringUtil.getMidString(ctx.channel().remoteAddress() + "", "/", ":"); |
|
||||||
// 截取端口号
|
|
||||||
port = ExchangeStringUtil.getMidString(ctx.channel().remoteAddress() + "", ":", ""); |
|
||||||
log.info("IP: " + IP + ",端口号: " + port); |
|
||||||
// 生成对应采集冷量计的命令
|
|
||||||
// 生成对应的采集指令
|
|
||||||
deviceManageEntityList = deviceManageService.queryDevicesByType("3"); |
|
||||||
size = deviceManageEntityList.size(); |
|
||||||
|
|
||||||
// 封装工具类进行采集,update by ljf on 2021-01-26
|
|
||||||
SendOrderUtils.sendCloudOrder(deviceManageEntityList.get(0),0,IP,port,ctx); |
|
||||||
// // 1.创建将要写出的数据
|
|
||||||
// String collectionNum = deviceManageEntityList.get(0).getCollectionNum();
|
|
||||||
// String sendStr = GetReadOrder485.createCloudOrder(IP, port,
|
|
||||||
// deviceManageEntityList.get(0).getDataCom(),
|
|
||||||
// collectionNum, "34");
|
|
||||||
//// String sendStr = "5803004900021914";
|
|
||||||
// ByteBuf buffer = getByteBuf(ctx, sendStr);
|
|
||||||
// // 2.发送数据
|
|
||||||
// ctx.channel().writeAndFlush(buffer);
|
|
||||||
} |
|
||||||
|
|
||||||
// String sendStr = "5803004900021914"; // 冷量计
|
|
||||||
// // 申请一个数据结构存储信息
|
|
||||||
// ByteBuf buffer = ctx.alloc().buffer();
|
|
||||||
// // 将信息放入数据结构中
|
|
||||||
// buffer.writeBytes(ExchangeStringUtil.hexStrToBinaryStr(sendStr));//对接需要16进制
|
|
||||||
// ctx.writeAndFlush(buffer, ctx.newProgressivePromise());
|
|
||||||
} |
|
||||||
|
|
||||||
private ByteBuf getByteBuf(ChannelHandlerContext ctx, String sendStr) { |
|
||||||
// byte类型的数据
|
|
||||||
// byte[] bytes = "这里是将要写往服务端的数据".getBytes(Charset.forName("utf-8"));
|
|
||||||
// String sendStr = "5803004900021914"; // 冷量计
|
|
||||||
// 申请一个数据结构存储信息
|
|
||||||
ByteBuf buffer = ctx.alloc().buffer(); |
|
||||||
// 将信息放入数据结构中
|
|
||||||
buffer.writeBytes(ExchangeStringUtil.hexStrToBinaryStr(sendStr));//对接需要16进制
|
|
||||||
return buffer; |
|
||||||
} |
|
||||||
|
|
||||||
@Override |
|
||||||
public void channelInactive(ChannelHandlerContext ctx) throws Exception { |
|
||||||
Thread.sleep(100); |
|
||||||
ctx.close(); |
|
||||||
log.info(ctx.channel().localAddress() + "退出链接!!"); |
|
||||||
} |
|
||||||
|
|
||||||
@Override |
|
||||||
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { |
|
||||||
try { |
|
||||||
ByteBuf buf = (ByteBuf)msg; |
|
||||||
byte [] bytes = new byte[buf.readableBytes()]; |
|
||||||
buf.readBytes(bytes);//复制内容到字节数组bytes
|
|
||||||
buf.clear(); |
|
||||||
log.info("获取到的值: " + ExchangeStringUtil.bytesToHexString(bytes)); |
|
||||||
if (bytes.length <= 36) { |
|
||||||
// receiveStr = receiveStr.replace("null", "");
|
|
||||||
// receiveStr = receiveStr + ExchangeStringUtil.bytesToHexString(bytes);//将接收到的数据转为字符串,此字符串就是客户端发送的字符串
|
|
||||||
// log.info(ctx.channel().remoteAddress() + " " + ctx.channel().localAddress() + " 接受服务器数据:" + receiveStr + ",大小: " + receiveStr.length());
|
|
||||||
receiveStr = receiveStr + ExchangeStringUtil.bytesToHexString(bytes);//将接收到的数据转为字符串,此字符串就是客户端发送的字符串
|
|
||||||
receiveStr = receiveStr.replace("null", ""); |
|
||||||
log.info("接受服务器数据:" + receiveStr + ",大小: " + receiveStr.length()); |
|
||||||
} |
|
||||||
} catch (Exception e) { |
|
||||||
e.printStackTrace(); |
|
||||||
} finally { |
|
||||||
ReferenceCountUtil.release(msg); |
|
||||||
} |
|
||||||
} |
|
||||||
|
|
||||||
@Override |
|
||||||
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { |
|
||||||
log.info("采集冷量计-数据读取接收完成: " + receiveStr); |
|
||||||
// A9 FE C2 C7 1F 90 01 58 03 04 4A 30 00 53 65 1C C4 06
|
|
||||||
if (receiveStr.length() == 36) { |
|
||||||
// 接收到的报文
|
|
||||||
log.info("接收完整报文: " + receiveStr); |
|
||||||
// 解析报文
|
|
||||||
// AnalysisReceiveOrder485 analysisReceiveOrder485 = new AnalysisReceiveOrder485();
|
|
||||||
// analysisReceiveOrder485.analysisCloudOrder485(receiveStr); // 解析冷量计
|
|
||||||
receiveStr = ""; |
|
||||||
// 1.创建将要写出的数据
|
|
||||||
// String sendStr = "5803004900021914";
|
|
||||||
num = num + 1; |
|
||||||
Thread.sleep(500); |
|
||||||
if (num > size-1) { |
|
||||||
num = 0; |
|
||||||
// 关闭连接
|
|
||||||
receiveStr = null; |
|
||||||
ctx.close(); |
|
||||||
// 保持长连接
|
|
||||||
// // 封装工具类进行采集,update by ljf on 2021-01-26
|
|
||||||
// SendOrderUtils.sendCloudOrder(deviceManageEntityList.get(num),num,IP,port,ctx);
|
|
||||||
} else { |
|
||||||
// 添加一个状态值,判断是否继续发送指令 update by ljf on 2020-08-07
|
|
||||||
if (Constant.WEB_FLAG) { |
|
||||||
log.info("有指令下发退出定时采集DDC参数"); |
|
||||||
num = 0; |
|
||||||
// 关闭连接
|
|
||||||
receiveStr = null; |
|
||||||
ctx.close(); |
|
||||||
} else { |
|
||||||
// 封装工具类进行采集,update by ljf on 2021-01-26
|
|
||||||
SendOrderUtils.sendCloudOrder(deviceManageEntityList.get(num),num,IP,port,ctx); |
|
||||||
// // 1.创建将要写出的数据
|
|
||||||
// String collectionNum = deviceManageEntityList.get(num).getCollectionNum();
|
|
||||||
// String sendStr = GetReadOrder485.createCloudOrder(IP, port,
|
|
||||||
// deviceManageEntityList.get(num).getDataCom(),
|
|
||||||
// collectionNum, "34");
|
|
||||||
//// String sendStr = "5803004900021914";
|
|
||||||
// ByteBuf buffer = getByteBuf(ctx, sendStr);
|
|
||||||
// // 2.发送数据
|
|
||||||
// ctx.channel().writeAndFlush(buffer);
|
|
||||||
// log.info("客户端再次往服务端发送数据" + sendStr);
|
|
||||||
} |
|
||||||
} |
|
||||||
|
|
||||||
} else { |
|
||||||
log.info(receiveStr); |
|
||||||
receiveStr = null; |
|
||||||
ctx.flush(); |
|
||||||
ctx.close(); |
|
||||||
} |
|
||||||
|
|
||||||
// if (receiveStr.contains("c0a801fc")) { // 冷量计
|
|
||||||
//
|
//
|
||||||
|
// @Override
|
||||||
|
// public void channelActive(ChannelHandlerContext ctx) {
|
||||||
|
// // 添加一个状态值,判断是否继续发送指令 update by ljf on 2020-08-07
|
||||||
|
// if (Constant.WEB_FLAG) {
|
||||||
|
// num = 0;
|
||||||
|
// // 关闭连接
|
||||||
|
// receiveStr = null;
|
||||||
|
// ctx.close();
|
||||||
|
// } else {
|
||||||
|
// ctx.channel().read();
|
||||||
|
// SimpleDateFormat sdf1 = new SimpleDateFormat("yyyy-mm-dd HH:mm:ss");
|
||||||
|
// Date date = new Date();
|
||||||
|
// log.info(ctx.channel().remoteAddress() + " " + sdf1.format(date) + "链接服务端成功!");
|
||||||
|
// // 截取IP地址
|
||||||
|
// IP = ExchangeStringUtil.getMidString(ctx.channel().remoteAddress() + "", "/", ":");
|
||||||
|
// // 截取端口号
|
||||||
|
// port = ExchangeStringUtil.getMidString(ctx.channel().remoteAddress() + "", ":", "");
|
||||||
|
// log.info("IP: " + IP + ",端口号: " + port);
|
||||||
|
// // 生成对应采集冷量计的命令
|
||||||
// // 生成对应的采集指令
|
// // 生成对应的采集指令
|
||||||
// deviceManageEntityList = deviceManageService.queryDevicesByType("3");
|
// deviceManageEntityList = deviceManageService.queryDevicesByType("3");
|
||||||
// size = deviceManageEntityList.size();
|
// size = deviceManageEntityList.size();
|
||||||
//
|
//
|
||||||
// log.info("初始连接报文: " + receiveStr);
|
// // 封装工具类进行采集,update by ljf on 2021-01-26
|
||||||
// IP = receiveStr;
|
// SendOrderUtils.sendCloudOrder(deviceManageEntityList.get(0),0,IP,port,ctx);
|
||||||
// receiveStr = "";
|
//// // 1.创建将要写出的数据
|
||||||
// // 1.创建将要写出的数据
|
//// String collectionNum = deviceManageEntityList.get(0).getCollectionNum();
|
||||||
// String collectionNum = deviceManageEntityList.get(0).getCollectionNum();
|
//// String sendStr = GetReadOrder485.createCloudOrder(IP, port,
|
||||||
// String sendStr = GetReadOrder485.createCloudOrder(collectionNum,"34");
|
//// deviceManageEntityList.get(0).getDataCom(),
|
||||||
//// String sendStr = "5803004900021914";
|
//// collectionNum, "34");
|
||||||
// ByteBuf buffer = getByteBuf(ctx,sendStr);
|
////// String sendStr = "5803004900021914";
|
||||||
// // 2.发送数据
|
//// ByteBuf buffer = getByteBuf(ctx, sendStr);
|
||||||
// ctx.channel().writeAndFlush(buffer);
|
//// // 2.发送数据
|
||||||
// } else if (receiveStr.contains("c0a801f0")) { // 电表
|
//// ctx.channel().writeAndFlush(buffer);
|
||||||
|
// }
|
||||||
//
|
//
|
||||||
// // 生成对应的采集指令
|
//// String sendStr = "5803004900021914"; // 冷量计
|
||||||
// deviceManageEntityList = deviceManageService.queryDevicesByType("1");
|
//// // 申请一个数据结构存储信息
|
||||||
// size = deviceManageEntityList.size();
|
//// ByteBuf buffer = ctx.alloc().buffer();
|
||||||
|
//// // 将信息放入数据结构中
|
||||||
|
//// buffer.writeBytes(ExchangeStringUtil.hexStrToBinaryStr(sendStr));//对接需要16进制
|
||||||
|
//// ctx.writeAndFlush(buffer, ctx.newProgressivePromise());
|
||||||
|
// }
|
||||||
//
|
//
|
||||||
// log.info("初始连接报文: " + receiveStr);
|
// private ByteBuf getByteBuf(ChannelHandlerContext ctx, String sendStr) {
|
||||||
// IP = receiveStr;
|
// // byte类型的数据
|
||||||
// receiveStr = "";
|
//// byte[] bytes = "这里是将要写往服务端的数据".getBytes(Charset.forName("utf-8"));
|
||||||
// // 1.创建将要写出的数据
|
//// String sendStr = "5803004900021914"; // 冷量计
|
||||||
//// String sendStr = "6830043080000068110432326536C816"; // 网络单相电表
|
// // 申请一个数据结构存储信息
|
||||||
//// String sendStr = "FEFEFEFE6880025007000068010243C3B216"; // 广仪三相电表
|
// ByteBuf buffer = ctx.alloc().buffer();
|
||||||
// String collectionNum = deviceManageEntityList.get(0).getCollectionNum();
|
// // 将信息放入数据结构中
|
||||||
// String sendStr = GetReadOrder485.createMeterOrder(collectionNum,"1");
|
// buffer.writeBytes(ExchangeStringUtil.hexStrToBinaryStr(sendStr));//对接需要16进制
|
||||||
// ByteBuf buffer = getByteBuf(ctx,sendStr);
|
// return buffer;
|
||||||
// // 2.发送数据
|
// }
|
||||||
// ctx.channel().writeAndFlush(buffer);
|
//
|
||||||
// } else if ((receiveStr.length() == 18) && (IP.contains("c0a801fc"))) {
|
// @Override
|
||||||
// analysisReceiveOrder485.analysisCloudOrder485(receiveStr); // 解析冷量计
|
// public void channelInactive(ChannelHandlerContext ctx) throws Exception {
|
||||||
|
// Thread.sleep(100);
|
||||||
|
// ctx.close();
|
||||||
|
// log.info(ctx.channel().localAddress() + "退出链接!!");
|
||||||
|
// }
|
||||||
|
//
|
||||||
|
// @Override
|
||||||
|
// public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
|
||||||
|
// try {
|
||||||
|
// ByteBuf buf = (ByteBuf)msg;
|
||||||
|
// byte [] bytes = new byte[buf.readableBytes()];
|
||||||
|
// buf.readBytes(bytes);//复制内容到字节数组bytes
|
||||||
|
// buf.clear();
|
||||||
|
// log.info("获取到的值: " + ExchangeStringUtil.bytesToHexString(bytes));
|
||||||
|
// if (bytes.length <= 36) {
|
||||||
|
//// receiveStr = receiveStr.replace("null", "");
|
||||||
|
//// receiveStr = receiveStr + ExchangeStringUtil.bytesToHexString(bytes);//将接收到的数据转为字符串,此字符串就是客户端发送的字符串
|
||||||
|
//// log.info(ctx.channel().remoteAddress() + " " + ctx.channel().localAddress() + " 接受服务器数据:" + receiveStr + ",大小: " + receiveStr.length());
|
||||||
|
// receiveStr = receiveStr + ExchangeStringUtil.bytesToHexString(bytes);//将接收到的数据转为字符串,此字符串就是客户端发送的字符串
|
||||||
|
// receiveStr = receiveStr.replace("null", "");
|
||||||
|
// log.info("接受服务器数据:" + receiveStr + ",大小: " + receiveStr.length());
|
||||||
|
// }
|
||||||
|
// } catch (Exception e) {
|
||||||
|
// e.printStackTrace();
|
||||||
|
// } finally {
|
||||||
|
// ReferenceCountUtil.release(msg);
|
||||||
|
// }
|
||||||
|
// }
|
||||||
|
//
|
||||||
|
// @Override
|
||||||
|
// public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
|
||||||
|
// log.info("采集冷量计-数据读取接收完成: " + receiveStr);
|
||||||
|
//// A9 FE C2 C7 1F 90 01 58 03 04 4A 30 00 53 65 1C C4 06
|
||||||
|
// if (receiveStr.length() == 36) {
|
||||||
|
// // 接收到的报文
|
||||||
|
// log.info("接收完整报文: " + receiveStr);
|
||||||
|
// // 解析报文
|
||||||
|
//// AnalysisReceiveOrder485 analysisReceiveOrder485 = new AnalysisReceiveOrder485();
|
||||||
|
//// analysisReceiveOrder485.analysisCloudOrder485(receiveStr); // 解析冷量计
|
||||||
// receiveStr = "";
|
// receiveStr = "";
|
||||||
// // 1.创建将要写出的数据
|
// // 1.创建将要写出的数据
|
||||||
//// String sendStr = "5803004900021914";
|
//// String sendStr = "5803004900021914";
|
||||||
// num = num + 1;
|
// num = num + 1;
|
||||||
// Thread.sleep(1000);
|
// Thread.sleep(500);
|
||||||
// if (num >= size-1) {
|
// if (num > size-1) {
|
||||||
// num = 0;
|
// num = 0;
|
||||||
// // 关闭连接
|
// // 关闭连接
|
||||||
// receiveStr = null;
|
// receiveStr = null;
|
||||||
// ctx.close();
|
// ctx.close();
|
||||||
|
//// 保持长连接
|
||||||
|
//// // 封装工具类进行采集,update by ljf on 2021-01-26
|
||||||
|
//// SendOrderUtils.sendCloudOrder(deviceManageEntityList.get(num),num,IP,port,ctx);
|
||||||
// } else {
|
// } else {
|
||||||
// String collectionNum = deviceManageEntityList.get(num).getCollectionNum();
|
// // 添加一个状态值,判断是否继续发送指令 update by ljf on 2020-08-07
|
||||||
// String sendStr = GetReadOrder485.createCloudOrder(collectionNum, "34");
|
// if (Constant.WEB_FLAG) {
|
||||||
// ByteBuf buffer = getByteBuf(ctx, sendStr);
|
// log.info("有指令下发退出定时采集DDC参数");
|
||||||
// // 2.发送数据
|
|
||||||
// ctx.channel().writeAndFlush(buffer);
|
|
||||||
// log.info("客户端再次往服务端发送数据" + num);
|
|
||||||
// }
|
|
||||||
//
|
|
||||||
// } else if ((receiveStr.length() == 44) && (IP.contains("c0a801f0"))) {
|
|
||||||
// analysisReceiveOrder485.analysisMeterOrder485(receiveStr); // 解析电表
|
|
||||||
// receiveStr = "";
|
|
||||||
// num = num + 1;
|
|
||||||
// Thread.sleep(1000);
|
|
||||||
// if (num >= size-1) {
|
|
||||||
// num = 0;
|
// num = 0;
|
||||||
// receiveStr = null;
|
|
||||||
// // 关闭连接
|
// // 关闭连接
|
||||||
|
// receiveStr = null;
|
||||||
// ctx.close();
|
// ctx.close();
|
||||||
// } else {
|
// } else {
|
||||||
// // 1.创建将要写出的数据
|
// // 封装工具类进行采集,update by ljf on 2021-01-26
|
||||||
// // fe fe fe fe 68 80 02 50 07 00 00 68 81 06 43 c3 8c 34 33 33 5c 16
|
// SendOrderUtils.sendCloudOrder(deviceManageEntityList.get(num),num,IP,port,ctx);
|
||||||
//// String sendStr = "FEFEFE6880025007000068010243C3B216";
|
//// // 1.创建将要写出的数据
|
||||||
// String collectionNum = deviceManageEntityList.get(num).getCollectionNum();
|
//// String collectionNum = deviceManageEntityList.get(num).getCollectionNum();
|
||||||
// String sendStr = GetReadOrder485.createMeterOrder(collectionNum, "1");
|
//// String sendStr = GetReadOrder485.createCloudOrder(IP, port,
|
||||||
// ByteBuf buffer = getByteBuf(ctx, sendStr);
|
//// deviceManageEntityList.get(num).getDataCom(),
|
||||||
// // 2.发送数据
|
//// collectionNum, "34");
|
||||||
// ctx.channel().writeAndFlush(buffer);
|
////// String sendStr = "5803004900021914";
|
||||||
// log.info("客户端再次往服务端发送数据" + num);
|
//// ByteBuf buffer = getByteBuf(ctx, sendStr);
|
||||||
|
//// // 2.发送数据
|
||||||
|
//// ctx.channel().writeAndFlush(buffer);
|
||||||
|
//// log.info("客户端再次往服务端发送数据" + sendStr);
|
||||||
|
// }
|
||||||
// }
|
// }
|
||||||
// } else if ((receiveStr.length() > 44)) {
|
//
|
||||||
|
// } else {
|
||||||
// log.info(receiveStr);
|
// log.info(receiveStr);
|
||||||
// receiveStr = null;
|
// receiveStr = null;
|
||||||
// ctx.flush();
|
// ctx.flush();
|
||||||
// ctx.close();
|
// ctx.close();
|
||||||
// }
|
// }
|
||||||
ctx.flush(); |
//
|
||||||
} |
//// if (receiveStr.contains("c0a801fc")) { // 冷量计
|
||||||
|
////
|
||||||
} |
//// // 生成对应的采集指令
|
||||||
|
//// deviceManageEntityList = deviceManageService.queryDevicesByType("3");
|
||||||
|
//// size = deviceManageEntityList.size();
|
||||||
|
////
|
||||||
|
//// log.info("初始连接报文: " + receiveStr);
|
||||||
|
//// IP = receiveStr;
|
||||||
|
//// receiveStr = "";
|
||||||
|
//// // 1.创建将要写出的数据
|
||||||
|
//// String collectionNum = deviceManageEntityList.get(0).getCollectionNum();
|
||||||
|
//// String sendStr = GetReadOrder485.createCloudOrder(collectionNum,"34");
|
||||||
|
////// String sendStr = "5803004900021914";
|
||||||
|
//// ByteBuf buffer = getByteBuf(ctx,sendStr);
|
||||||
|
//// // 2.发送数据
|
||||||
|
//// ctx.channel().writeAndFlush(buffer);
|
||||||
|
//// } else if (receiveStr.contains("c0a801f0")) { // 电表
|
||||||
|
////
|
||||||
|
//// // 生成对应的采集指令
|
||||||
|
//// deviceManageEntityList = deviceManageService.queryDevicesByType("1");
|
||||||
|
//// size = deviceManageEntityList.size();
|
||||||
|
////
|
||||||
|
//// log.info("初始连接报文: " + receiveStr);
|
||||||
|
//// IP = receiveStr;
|
||||||
|
//// receiveStr = "";
|
||||||
|
//// // 1.创建将要写出的数据
|
||||||
|
////// String sendStr = "6830043080000068110432326536C816"; // 网络单相电表
|
||||||
|
////// String sendStr = "FEFEFEFE6880025007000068010243C3B216"; // 广仪三相电表
|
||||||
|
//// String collectionNum = deviceManageEntityList.get(0).getCollectionNum();
|
||||||
|
//// String sendStr = GetReadOrder485.createMeterOrder(collectionNum,"1");
|
||||||
|
//// ByteBuf buffer = getByteBuf(ctx,sendStr);
|
||||||
|
//// // 2.发送数据
|
||||||
|
//// ctx.channel().writeAndFlush(buffer);
|
||||||
|
//// } else if ((receiveStr.length() == 18) && (IP.contains("c0a801fc"))) {
|
||||||
|
//// analysisReceiveOrder485.analysisCloudOrder485(receiveStr); // 解析冷量计
|
||||||
|
//// receiveStr = "";
|
||||||
|
//// // 1.创建将要写出的数据
|
||||||
|
////// String sendStr = "5803004900021914";
|
||||||
|
//// num = num + 1;
|
||||||
|
//// Thread.sleep(1000);
|
||||||
|
//// if (num >= size-1) {
|
||||||
|
//// num = 0;
|
||||||
|
//// // 关闭连接
|
||||||
|
//// receiveStr = null;
|
||||||
|
//// ctx.close();
|
||||||
|
//// } else {
|
||||||
|
//// String collectionNum = deviceManageEntityList.get(num).getCollectionNum();
|
||||||
|
//// String sendStr = GetReadOrder485.createCloudOrder(collectionNum, "34");
|
||||||
|
//// ByteBuf buffer = getByteBuf(ctx, sendStr);
|
||||||
|
//// // 2.发送数据
|
||||||
|
//// ctx.channel().writeAndFlush(buffer);
|
||||||
|
//// log.info("客户端再次往服务端发送数据" + num);
|
||||||
|
//// }
|
||||||
|
////
|
||||||
|
//// } else if ((receiveStr.length() == 44) && (IP.contains("c0a801f0"))) {
|
||||||
|
//// analysisReceiveOrder485.analysisMeterOrder485(receiveStr); // 解析电表
|
||||||
|
//// receiveStr = "";
|
||||||
|
//// num = num + 1;
|
||||||
|
//// Thread.sleep(1000);
|
||||||
|
//// if (num >= size-1) {
|
||||||
|
//// num = 0;
|
||||||
|
//// receiveStr = null;
|
||||||
|
//// // 关闭连接
|
||||||
|
//// ctx.close();
|
||||||
|
//// } else {
|
||||||
|
//// // 1.创建将要写出的数据
|
||||||
|
//// // fe fe fe fe 68 80 02 50 07 00 00 68 81 06 43 c3 8c 34 33 33 5c 16
|
||||||
|
////// String sendStr = "FEFEFE6880025007000068010243C3B216";
|
||||||
|
//// String collectionNum = deviceManageEntityList.get(num).getCollectionNum();
|
||||||
|
//// String sendStr = GetReadOrder485.createMeterOrder(collectionNum, "1");
|
||||||
|
//// ByteBuf buffer = getByteBuf(ctx, sendStr);
|
||||||
|
//// // 2.发送数据
|
||||||
|
//// ctx.channel().writeAndFlush(buffer);
|
||||||
|
//// log.info("客户端再次往服务端发送数据" + num);
|
||||||
|
//// }
|
||||||
|
//// } else if ((receiveStr.length() > 44)) {
|
||||||
|
//// log.info(receiveStr);
|
||||||
|
//// receiveStr = null;
|
||||||
|
//// ctx.flush();
|
||||||
|
//// ctx.close();
|
||||||
|
//// }
|
||||||
|
// ctx.flush();
|
||||||
|
// }
|
||||||
|
//
|
||||||
|
//}
|
||||||
|
@ -1,179 +1,179 @@ |
|||||||
package com.mh.user.netty; |
//package com.mh.user.netty;
|
||||||
|
//
|
||||||
import com.mh.user.utils.ExchangeStringUtil; |
//import com.mh.user.utils.ExchangeStringUtil;
|
||||||
import io.netty.bootstrap.ServerBootstrap; |
//import io.netty.bootstrap.ServerBootstrap;
|
||||||
import io.netty.buffer.ByteBuf; |
//import io.netty.buffer.ByteBuf;
|
||||||
import io.netty.buffer.Unpooled; |
//import io.netty.buffer.Unpooled;
|
||||||
import io.netty.channel.*; |
//import io.netty.channel.*;
|
||||||
import io.netty.channel.nio.NioEventLoopGroup; |
//import io.netty.channel.nio.NioEventLoopGroup;
|
||||||
import io.netty.channel.socket.SocketChannel; |
//import io.netty.channel.socket.SocketChannel;
|
||||||
import io.netty.channel.socket.nio.NioServerSocketChannel; |
//import io.netty.channel.socket.nio.NioServerSocketChannel;
|
||||||
import io.netty.handler.logging.LogLevel; |
//import io.netty.handler.logging.LogLevel;
|
||||||
import io.netty.handler.logging.LoggingHandler; |
//import io.netty.handler.logging.LoggingHandler;
|
||||||
import lombok.extern.slf4j.Slf4j; |
//import lombok.extern.slf4j.Slf4j;
|
||||||
import org.springframework.util.StringUtils; |
//import org.springframework.util.StringUtils;
|
||||||
|
//
|
||||||
import java.io.IOException; |
//import java.io.IOException;
|
||||||
import java.net.InetSocketAddress; |
//import java.net.InetSocketAddress;
|
||||||
|
//
|
||||||
|
//
|
||||||
/** |
///**
|
||||||
* @author ljf |
// * @author ljf
|
||||||
* @title :Netty |
// * @title :Netty
|
||||||
* @description :netty 使用 |
// * @description :netty 使用
|
||||||
* @updateTime 2020-04-21 |
// * @updateTime 2020-04-21
|
||||||
* @throws : |
// * @throws :
|
||||||
*/ |
// */
|
||||||
@Slf4j |
//@Slf4j
|
||||||
public class NettyEchoServer { |
//public class NettyEchoServer {
|
||||||
|
//
|
||||||
public void bind(int port) throws Exception { |
// public void bind(int port) throws Exception {
|
||||||
// accept线程组,用来接收连接
|
// // accept线程组,用来接收连接
|
||||||
EventLoopGroup bossGroup = new NioEventLoopGroup(1); |
// EventLoopGroup bossGroup = new NioEventLoopGroup(1);
|
||||||
// IO 线程组,用来处理业务逻辑
|
// // IO 线程组,用来处理业务逻辑
|
||||||
EventLoopGroup workerGroup = new NioEventLoopGroup(1); |
// EventLoopGroup workerGroup = new NioEventLoopGroup(1);
|
||||||
|
//
|
||||||
try { |
// try {
|
||||||
// 服务端启动引导
|
// // 服务端启动引导
|
||||||
ServerBootstrap serverBootstrap = new ServerBootstrap(); |
// ServerBootstrap serverBootstrap = new ServerBootstrap();
|
||||||
serverBootstrap.group(bossGroup,workerGroup) // 绑定两个线程
|
// serverBootstrap.group(bossGroup,workerGroup) // 绑定两个线程
|
||||||
.channel(NioServerSocketChannel.class) // 指定通道类型
|
// .channel(NioServerSocketChannel.class) // 指定通道类型
|
||||||
.option(ChannelOption.SO_BACKLOG, 1024) // 设置TCP连接的缓冲区
|
// .option(ChannelOption.SO_BACKLOG, 1024) // 设置TCP连接的缓冲区
|
||||||
.handler(new LoggingHandler(LogLevel.INFO)) // 设置日志级别
|
// .handler(new LoggingHandler(LogLevel.INFO)) // 设置日志级别
|
||||||
.childHandler(new ChannelInitializer<SocketChannel>() { |
// .childHandler(new ChannelInitializer<SocketChannel>() {
|
||||||
@Override |
// @Override
|
||||||
protected void initChannel(SocketChannel socketChannel) throws Exception { |
// protected void initChannel(SocketChannel socketChannel) throws Exception {
|
||||||
ChannelPipeline pipeline = socketChannel.pipeline(); // 获取处理器链
|
// ChannelPipeline pipeline = socketChannel.pipeline(); // 获取处理器链
|
||||||
pipeline.addLast(new EchoServerHandler()); // 添加新的事件处理器
|
// pipeline.addLast(new EchoServerHandler()); // 添加新的事件处理器
|
||||||
} |
// }
|
||||||
}); |
// });
|
||||||
// 通过bind启动服务
|
// // 通过bind启动服务
|
||||||
ChannelFuture f = serverBootstrap.bind(port).sync(); |
// ChannelFuture f = serverBootstrap.bind(port).sync();
|
||||||
// 阻塞主线程,知道网络服务被关闭
|
// // 阻塞主线程,知道网络服务被关闭
|
||||||
f.channel().closeFuture().sync(); |
// f.channel().closeFuture().sync();
|
||||||
} catch (Exception e){ |
// } catch (Exception e){
|
||||||
e.printStackTrace(); |
// e.printStackTrace();
|
||||||
} finally { |
// } finally {
|
||||||
workerGroup.shutdownGracefully(); |
// workerGroup.shutdownGracefully();
|
||||||
bossGroup.shutdownGracefully(); |
// bossGroup.shutdownGracefully();
|
||||||
} |
// }
|
||||||
} |
// }
|
||||||
|
//
|
||||||
static class EchoServerHandler extends ChannelHandlerAdapter { |
// static class EchoServerHandler extends ChannelHandlerAdapter {
|
||||||
|
//
|
||||||
// 每当从客户端收到新的数据时,这个方法会在收到消息时被调用
|
// // 每当从客户端收到新的数据时,这个方法会在收到消息时被调用
|
||||||
@Override |
// @Override
|
||||||
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { |
// public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
|
||||||
try { |
// try {
|
||||||
ByteBuf buf = (ByteBuf)msg; |
// ByteBuf buf = (ByteBuf)msg;
|
||||||
byte [] bytes = new byte[buf.readableBytes()]; |
// byte [] bytes = new byte[buf.readableBytes()];
|
||||||
buf.readBytes(bytes);//复制内容到字节数组bytes
|
// buf.readBytes(bytes);//复制内容到字节数组bytes
|
||||||
String receiveStr = ExchangeStringUtil.bytesToHexString(bytes);//将接收到的数据转为字符串,此字符串就是客户端发送的字符串
|
// String receiveStr = ExchangeStringUtil.bytesToHexString(bytes);//将接收到的数据转为字符串,此字符串就是客户端发送的字符串
|
||||||
log.info("接收到的数据: "+ receiveStr); |
// log.info("接收到的数据: "+ receiveStr);
|
||||||
//返回16进制到客户端
|
// //返回16进制到客户端
|
||||||
writeToClient(receiveStr,ctx,"测试"); |
// writeToClient(receiveStr,ctx,"测试");
|
||||||
} catch (Exception e) { |
// } catch (Exception e) {
|
||||||
// TODO Auto-generated catch block
|
// // TODO Auto-generated catch block
|
||||||
e.printStackTrace(); |
// e.printStackTrace();
|
||||||
} |
// }
|
||||||
// ctx.write(Unpooled.wrappedBuffer("Server message".getBytes()));
|
//// ctx.write(Unpooled.wrappedBuffer("Server message".getBytes()));
|
||||||
// ctx.fireChannelRead(msg);
|
//// ctx.fireChannelRead(msg);
|
||||||
} |
// }
|
||||||
|
//
|
||||||
// 数据读取完后被调用
|
// // 数据读取完后被调用
|
||||||
@Override |
// @Override
|
||||||
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { |
// public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
|
||||||
ctx.flush(); |
// ctx.flush();
|
||||||
} |
// }
|
||||||
|
//
|
||||||
// 当Netty由于IO错误或者处理器在处理事件时抛出的异常时被调用
|
// // 当Netty由于IO错误或者处理器在处理事件时抛出的异常时被调用
|
||||||
@Override |
// @Override
|
||||||
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { |
// public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
|
||||||
cause.printStackTrace(); |
// cause.printStackTrace();
|
||||||
ctx.close(); |
// ctx.close();
|
||||||
} |
// }
|
||||||
|
//
|
||||||
/** |
// /**
|
||||||
* 客户端与服务端第一次建立连接时 执行 |
// * 客户端与服务端第一次建立连接时 执行
|
||||||
* |
// *
|
||||||
* @param ctx |
// * @param ctx
|
||||||
* @throws Exception |
// * @throws Exception
|
||||||
*/ |
// */
|
||||||
@Override |
// @Override
|
||||||
public void channelActive(ChannelHandlerContext ctx) throws Exception, IOException |
// public void channelActive(ChannelHandlerContext ctx) throws Exception, IOException
|
||||||
{ |
// {
|
||||||
super.channelActive(ctx); |
// super.channelActive(ctx);
|
||||||
ctx.channel().read(); |
// ctx.channel().read();
|
||||||
InetSocketAddress ifsock = (InetSocketAddress) ctx.channel().remoteAddress(); |
// InetSocketAddress ifsock = (InetSocketAddress) ctx.channel().remoteAddress();
|
||||||
String clientIp = ifsock.getAddress().getHostAddress(); |
// String clientIp = ifsock.getAddress().getHostAddress();
|
||||||
//此处不能使用ctx.close(),否则客户端始终无法与服务端建立连接
|
// //此处不能使用ctx.close(),否则客户端始终无法与服务端建立连接
|
||||||
log.info("channelActive: "+clientIp + ctx.name()); |
// log.info("channelActive: "+clientIp + ctx.name());
|
||||||
} |
// }
|
||||||
|
//
|
||||||
/** |
// /**
|
||||||
* 客户端与服务端 断连时 执行 |
// * 客户端与服务端 断连时 执行
|
||||||
* |
// *
|
||||||
* @param ctx |
// * @param ctx
|
||||||
* @throws Exception |
// * @throws Exception
|
||||||
*/ |
// */
|
||||||
@Override |
// @Override
|
||||||
public void channelInactive(ChannelHandlerContext ctx) throws Exception, IOException |
// public void channelInactive(ChannelHandlerContext ctx) throws Exception, IOException
|
||||||
{ |
// {
|
||||||
super.channelInactive(ctx); |
// super.channelInactive(ctx);
|
||||||
InetSocketAddress ifsock = (InetSocketAddress) ctx.channel().remoteAddress(); |
// InetSocketAddress ifsock = (InetSocketAddress) ctx.channel().remoteAddress();
|
||||||
String clientIp = ifsock.getAddress().getHostAddress(); |
// String clientIp = ifsock.getAddress().getHostAddress();
|
||||||
ctx.close(); //断开连接时,必须关闭,否则造成资源浪费,并发量很大情况下可能造成宕机
|
// ctx.close(); //断开连接时,必须关闭,否则造成资源浪费,并发量很大情况下可能造成宕机
|
||||||
System.out.println("channelInactive:"+clientIp); |
// System.out.println("channelInactive:"+clientIp);
|
||||||
} |
// }
|
||||||
|
//
|
||||||
/** |
// /**
|
||||||
* 服务端当read超时, 会调用这个方法 |
// * 服务端当read超时, 会调用这个方法
|
||||||
* |
// *
|
||||||
* @param ctx |
// * @param ctx
|
||||||
* @param evt |
// * @param evt
|
||||||
* @throws Exception |
// * @throws Exception
|
||||||
*/ |
// */
|
||||||
@Override |
// @Override
|
||||||
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception, IOException |
// public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception, IOException
|
||||||
{ |
// {
|
||||||
super.userEventTriggered(ctx, evt); |
// super.userEventTriggered(ctx, evt);
|
||||||
InetSocketAddress ifsock = (InetSocketAddress) ctx.channel().remoteAddress(); |
// InetSocketAddress ifsock = (InetSocketAddress) ctx.channel().remoteAddress();
|
||||||
String clientIp = ifsock.getAddress().getHostAddress(); |
// String clientIp = ifsock.getAddress().getHostAddress();
|
||||||
ctx.close();//超时时断开连接
|
// ctx.close();//超时时断开连接
|
||||||
System.out.println("userEventTriggered:"+clientIp); |
// System.out.println("userEventTriggered:"+clientIp);
|
||||||
} |
// }
|
||||||
|
//
|
||||||
|
//
|
||||||
/** |
// /**
|
||||||
* 公用回写数据到客户端的方法 |
// * 公用回写数据到客户端的方法
|
||||||
* @param channel |
// * @param channel
|
||||||
* @param mark 用于打印/log的输出 |
// * @param mark 用于打印/log的输出
|
||||||
* <br>//channel.writeAndFlush(msg);//不行
|
// * <br>//channel.writeAndFlush(msg);//不行
|
||||||
* <br>//channel.writeAndFlush(receiveStr.getBytes());//不行
|
// * <br>//channel.writeAndFlush(receiveStr.getBytes());//不行
|
||||||
* <br>在netty里,进出的都是ByteBuf,楼主应确定服务端是否有对应的编码器,将字符串转化为ByteBuf |
// * <br>在netty里,进出的都是ByteBuf,楼主应确定服务端是否有对应的编码器,将字符串转化为ByteBuf
|
||||||
*/ |
// */
|
||||||
private void writeToClient(final String receiveStr, ChannelHandlerContext channel, final String mark) { |
// private void writeToClient(final String receiveStr, ChannelHandlerContext channel, final String mark) {
|
||||||
try { |
// try {
|
||||||
ByteBuf buff = Unpooled.buffer();//netty需要用ByteBuf传输
|
// ByteBuf buff = Unpooled.buffer();//netty需要用ByteBuf传输
|
||||||
buff.writeBytes(ExchangeStringUtil.hexStrToBinaryStr(receiveStr));//对接需要16进制
|
// buff.writeBytes(ExchangeStringUtil.hexStrToBinaryStr(receiveStr));//对接需要16进制
|
||||||
channel.writeAndFlush(buff).addListener((ChannelFutureListener) future -> { |
// channel.writeAndFlush(buff).addListener((ChannelFutureListener) future -> {
|
||||||
StringBuilder sb = new StringBuilder(""); |
// StringBuilder sb = new StringBuilder("");
|
||||||
if(!StringUtils.isEmpty(mark)){ |
// if(!StringUtils.isEmpty(mark)){
|
||||||
sb.append("【").append(mark).append("】"); |
// sb.append("【").append(mark).append("】");
|
||||||
} |
// }
|
||||||
if (future.isSuccess()) { |
// if (future.isSuccess()) {
|
||||||
System.out.println(sb.toString()+"回写成功"+receiveStr); |
// System.out.println(sb.toString()+"回写成功"+receiveStr);
|
||||||
log.info(sb.toString()+"回写成功"+receiveStr); |
// log.info(sb.toString()+"回写成功"+receiveStr);
|
||||||
} else { |
// } else {
|
||||||
System.out.println(sb.toString()+"回写失败"+receiveStr); |
// System.out.println(sb.toString()+"回写失败"+receiveStr);
|
||||||
log.error(sb.toString()+"回写失败"+receiveStr); |
// log.error(sb.toString()+"回写失败"+receiveStr);
|
||||||
} |
// }
|
||||||
}); |
// });
|
||||||
} catch (Exception e) { |
// } catch (Exception e) {
|
||||||
e.printStackTrace(); |
// e.printStackTrace();
|
||||||
System.out.println("调用通用writeToClient()异常"+e.getMessage()); |
// System.out.println("调用通用writeToClient()异常"+e.getMessage());
|
||||||
log.error("调用通用writeToClient()异常:",e); |
// log.error("调用通用writeToClient()异常:",e);
|
||||||
} |
// }
|
||||||
} |
// }
|
||||||
} |
// }
|
||||||
} |
//}
|
||||||
|
@ -1,193 +1,193 @@ |
|||||||
package com.mh.user.netty; |
//package com.mh.user.netty;
|
||||||
|
//
|
||||||
import com.mh.user.constants.Constant; |
//import com.mh.user.constants.Constant;
|
||||||
import com.mh.user.entity.DeviceManageEntity; |
//import com.mh.user.entity.DeviceManageEntity;
|
||||||
import com.mh.user.service.DeviceManageService; |
//import com.mh.user.service.DeviceManageService;
|
||||||
import com.mh.user.utils.*; |
//import com.mh.user.utils.*;
|
||||||
import io.netty.buffer.ByteBuf; |
//import io.netty.buffer.ByteBuf;
|
||||||
import io.netty.channel.Channel; |
//import io.netty.channel.Channel;
|
||||||
import io.netty.channel.ChannelHandlerAdapter; |
//import io.netty.channel.ChannelHandlerAdapter;
|
||||||
import io.netty.channel.ChannelHandlerContext; |
//import io.netty.channel.ChannelHandlerContext;
|
||||||
import io.netty.handler.timeout.IdleState; |
//import io.netty.handler.timeout.IdleState;
|
||||||
import io.netty.handler.timeout.IdleStateEvent; |
//import io.netty.handler.timeout.IdleStateEvent;
|
||||||
import io.netty.util.ReferenceCountUtil; |
//import io.netty.util.ReferenceCountUtil;
|
||||||
import lombok.extern.slf4j.Slf4j; |
//import lombok.extern.slf4j.Slf4j;
|
||||||
import org.springframework.context.ApplicationContext; |
//import org.springframework.context.ApplicationContext;
|
||||||
|
//
|
||||||
import java.text.SimpleDateFormat; |
//import java.text.SimpleDateFormat;
|
||||||
import java.util.Date; |
//import java.util.Date;
|
||||||
import java.util.List; |
//import java.util.List;
|
||||||
|
//
|
||||||
/** |
///**
|
||||||
* @author ljf |
// * @author ljf
|
||||||
* @title : |
// * @title :
|
||||||
* @description :客户端异步消息处理机制 |
// * @description :客户端异步消息处理机制
|
||||||
* @updateTime 2020-05-13 |
// * @updateTime 2020-05-13
|
||||||
* @throws : |
// * @throws :
|
||||||
*/ |
// */
|
||||||
@Slf4j |
//@Slf4j
|
||||||
public class NettyMeterAndCloudClientHandler extends ChannelHandlerAdapter { |
//public class NettyMeterAndCloudClientHandler extends ChannelHandlerAdapter {
|
||||||
|
//
|
||||||
|
//
|
||||||
private int num = 0; |
// private int num = 0;
|
||||||
private int size = 0; |
// private int size = 0;
|
||||||
private String receiveStr = null; |
// private String receiveStr = null;
|
||||||
private int idle_count = 0; |
// private int idle_count = 0;
|
||||||
private String IP = ""; |
// private String IP = "";
|
||||||
private String port = ""; |
// private String port = "";
|
||||||
|
//
|
||||||
List<DeviceManageEntity> deviceManageEntityList; |
// List<DeviceManageEntity> deviceManageEntityList;
|
||||||
|
//
|
||||||
// 调用service
|
// // 调用service
|
||||||
ApplicationContext context = SpringBeanUtil.getApplicationContext(); |
// ApplicationContext context = SpringBeanUtil.getApplicationContext();
|
||||||
DeviceManageService deviceManageService = context.getBean(DeviceManageService.class); |
// DeviceManageService deviceManageService = context.getBean(DeviceManageService.class);
|
||||||
|
//
|
||||||
AnalysisReceiveOrder485 analysisReceiveOrder485 = new AnalysisReceiveOrder485(); |
// AnalysisReceiveOrder485 analysisReceiveOrder485 = new AnalysisReceiveOrder485();
|
||||||
|
//
|
||||||
@Override |
// @Override
|
||||||
public void channelUnregistered(ChannelHandlerContext ctx) throws Exception { |
// public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
|
||||||
log.info("当前channel从EventLoop取消注册"); |
// log.info("当前channel从EventLoop取消注册");
|
||||||
super.channelUnregistered(ctx); |
// super.channelUnregistered(ctx);
|
||||||
} |
// }
|
||||||
|
//
|
||||||
/** |
// /**
|
||||||
* 超时处理 |
// * 超时处理
|
||||||
* 如果120秒没有接受客户端的心跳,就触发; |
// * 如果120秒没有接受客户端的心跳,就触发;
|
||||||
* 如果超过3次,则直接关闭; |
// * 如果超过3次,则直接关闭;
|
||||||
*/ |
// */
|
||||||
@Override |
// @Override
|
||||||
public void userEventTriggered(ChannelHandlerContext ctx, Object obj) throws Exception { |
// public void userEventTriggered(ChannelHandlerContext ctx, Object obj) throws Exception {
|
||||||
if (obj instanceof IdleStateEvent) { |
// if (obj instanceof IdleStateEvent) {
|
||||||
IdleStateEvent event = (IdleStateEvent) obj; |
// IdleStateEvent event = (IdleStateEvent) obj;
|
||||||
if (IdleState.READER_IDLE.equals(event.state())) { //如果读通道处于空闲状态,说明没有接收到心跳命令
|
// if (IdleState.READER_IDLE.equals(event.state())) { //如果读通道处于空闲状态,说明没有接收到心跳命令
|
||||||
System.out.println("第" + idle_count + "已经20秒没有接收到服务器的信息了,发送的第" + num + "条数据"); |
// System.out.println("第" + idle_count + "已经20秒没有接收到服务器的信息了,发送的第" + num + "条数据");
|
||||||
if (deviceManageEntityList.get(num) == null) { |
// if (deviceManageEntityList.get(num) == null) {
|
||||||
ctx.channel().close(); |
// ctx.channel().close();
|
||||||
} else { |
// } else {
|
||||||
if (idle_count > 3 || num > size - 1) { |
// if (idle_count > 3 || num > size - 1) {
|
||||||
System.out.println("关闭这个不活跃的channel"); |
// System.out.println("关闭这个不活跃的channel");
|
||||||
ctx.channel().close(); |
// ctx.channel().close();
|
||||||
} |
// }
|
||||||
|
//
|
||||||
SendOrderUtils.sendMeterOrCloud(deviceManageEntityList.get(num), num, IP, port, ctx); |
// SendOrderUtils.sendMeterOrCloud(deviceManageEntityList.get(num), num, IP, port, ctx);
|
||||||
idle_count++; |
// idle_count++;
|
||||||
} |
// }
|
||||||
} |
// }
|
||||||
} else { |
// } else {
|
||||||
super.userEventTriggered(ctx, obj); |
// super.userEventTriggered(ctx, obj);
|
||||||
} |
// }
|
||||||
} |
// }
|
||||||
|
//
|
||||||
@Override |
// @Override
|
||||||
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { |
// public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
|
||||||
log.info("通信异常!!"); |
// log.info("通信异常!!");
|
||||||
Channel incoming = ctx.channel(); |
// Channel incoming = ctx.channel();
|
||||||
if (incoming.isActive()) { |
// if (incoming.isActive()) {
|
||||||
log.info("SimpleClient: " + incoming.remoteAddress() + "异常"); |
// log.info("SimpleClient: " + incoming.remoteAddress() + "异常");
|
||||||
receiveStr = null; |
// receiveStr = null;
|
||||||
cause.printStackTrace(); |
// cause.printStackTrace();
|
||||||
ctx.close(); |
// ctx.close();
|
||||||
} |
// }
|
||||||
} |
// }
|
||||||
|
//
|
||||||
|
//
|
||||||
@Override |
// @Override
|
||||||
public void channelActive(ChannelHandlerContext ctx) throws Exception { |
// public void channelActive(ChannelHandlerContext ctx) throws Exception {
|
||||||
// super.channelActive(ctx);
|
//// super.channelActive(ctx);
|
||||||
// 添加一个状态值,判断是否继续发送指令 update by ljf on 2020-08-07
|
// // 添加一个状态值,判断是否继续发送指令 update by ljf on 2020-08-07
|
||||||
if (Constant.WEB_FLAG) { |
// if (Constant.WEB_FLAG) {
|
||||||
num = 0; |
// num = 0;
|
||||||
// 关闭连接
|
// // 关闭连接
|
||||||
receiveStr = null; |
// receiveStr = null;
|
||||||
ctx.close(); |
// ctx.close();
|
||||||
} else { |
// } else {
|
||||||
SimpleDateFormat sdf1 = new SimpleDateFormat("yyyy-mm-dd HH:mm:ss"); |
// SimpleDateFormat sdf1 = new SimpleDateFormat("yyyy-mm-dd HH:mm:ss");
|
||||||
Date date = new Date(); |
// Date date = new Date();
|
||||||
log.info(ctx.channel().remoteAddress() + " " + sdf1.format(date) + "链接服务端成功!"); |
// log.info(ctx.channel().remoteAddress() + " " + sdf1.format(date) + "链接服务端成功!");
|
||||||
|
//
|
||||||
// 截取IP地址
|
// // 截取IP地址
|
||||||
IP = ExchangeStringUtil.getMidString(ctx.channel().remoteAddress() + "", "/", ":"); |
// IP = ExchangeStringUtil.getMidString(ctx.channel().remoteAddress() + "", "/", ":");
|
||||||
// 截取端口号
|
// // 截取端口号
|
||||||
port = ExchangeStringUtil.getMidString(ctx.channel().remoteAddress() + "", ":", ""); |
// port = ExchangeStringUtil.getMidString(ctx.channel().remoteAddress() + "", ":", "");
|
||||||
log.info("IP: " + IP + ",端口号: " + port); |
// log.info("IP: " + IP + ",端口号: " + port);
|
||||||
|
//
|
||||||
// 生成对应的采集指令
|
// // 生成对应的采集指令
|
||||||
deviceManageEntityList = deviceManageService.queryDevicesByType(null); |
// deviceManageEntityList = deviceManageService.queryDevicesByType(null);
|
||||||
size = deviceManageEntityList.size(); |
// size = deviceManageEntityList.size();
|
||||||
|
//
|
||||||
log.info("初始连接报文: " + receiveStr); |
// log.info("初始连接报文: " + receiveStr);
|
||||||
receiveStr = ""; |
// receiveStr = "";
|
||||||
// 保持长连接,封装发送电表工具方法 update by ljf on 2021-01-26
|
|
||||||
SendOrderUtils.sendMeterOrCloud(deviceManageEntityList.get(0), 0, IP, port, ctx); |
|
||||||
} |
|
||||||
} |
|
||||||
|
|
||||||
@Override |
|
||||||
public void channelInactive(ChannelHandlerContext ctx) throws Exception { |
|
||||||
Thread.sleep(500); |
|
||||||
receiveStr = null; |
|
||||||
ctx.close(); |
|
||||||
log.info(ctx.channel().localAddress() + "退出链接!!"); |
|
||||||
} |
|
||||||
|
|
||||||
@Override |
|
||||||
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { |
|
||||||
try { |
|
||||||
ByteBuf buf = (ByteBuf) msg; |
|
||||||
byte[] bytes = new byte[buf.readableBytes()]; |
|
||||||
buf.readBytes(bytes);//复制内容到字节数组bytes
|
|
||||||
buf.clear(); |
|
||||||
log.info("获取到的值: " + ExchangeStringUtil.bytesToHexString(bytes)); |
|
||||||
if (bytes.length <= 62) { |
|
||||||
// if (bytes.length <= 142) {
|
|
||||||
// receiveStr = receiveStr.replace("null", "");
|
|
||||||
// receiveStr = receiveStr + ExchangeStringUtil.bytesToHexString(bytes);//将接收到的数据转为字符串,此字符串就是客户端发送的字符串
|
|
||||||
// log.info(ctx.channel().remoteAddress() + " " + ctx.channel().localAddress() + " 接受服务器数据:" + receiveStr + ",大小: " + receiveStr.length());
|
|
||||||
receiveStr = receiveStr + ExchangeStringUtil.bytesToHexString(bytes);//将接收到的数据转为字符串,此字符串就是客户端发送的字符串
|
|
||||||
receiveStr = receiveStr.replace("null", ""); |
|
||||||
log.info("接受服务器数据:" + receiveStr + ",大小: " + receiveStr.length()); |
|
||||||
} |
|
||||||
} catch (Exception e) { |
|
||||||
e.printStackTrace(); |
|
||||||
} finally { |
|
||||||
ReferenceCountUtil.release(msg); |
|
||||||
} |
|
||||||
} |
|
||||||
|
|
||||||
@Override |
|
||||||
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { |
|
||||||
log.info("采集电表或者冷量计-数据读取接收完成: " + receiveStr); |
|
||||||
if ((receiveStr.length() == 36) && (deviceManageEntityList.get(num).getParamId() == 3)) { |
|
||||||
// analysisReceiveOrder485.analysisCloudOrder485(receiveStr); // 解析冷量计
|
|
||||||
} else if ((receiveStr.length() == 62) && (deviceManageEntityList.get(num).getParamId() == 1)) { |
|
||||||
// 把receiveStr的"null"值去掉
|
|
||||||
// a9fec2c71f9002fefefefe6839025007000068810643c3bb446c338d16c2b8
|
|
||||||
// A9 FE C2 C7 1F 90 02 FE FE FE FE 68 39 02 50 07 00 00 68 81 06 43 C3 5B 38 6C 33 21 16 F8 12
|
|
||||||
analysisReceiveOrder485.analysisMeterOrder485(receiveStr,"","",""); // 解析电表
|
|
||||||
} |
|
||||||
receiveStr = ""; |
|
||||||
num = num + 1; |
|
||||||
Thread.sleep(600); |
|
||||||
if (num > size - 1) { |
|
||||||
num = 0; |
|
||||||
receiveStr = null; |
|
||||||
// 关闭连接
|
|
||||||
ctx.close(); |
|
||||||
// // 保持长连接,封装发送电表工具方法 update by ljf on 2021-01-26
|
// // 保持长连接,封装发送电表工具方法 update by ljf on 2021-01-26
|
||||||
// SendOrderUtils.sendMeterOrder(deviceManageEntityList.get(num),num,IP,port,ctx);
|
// SendOrderUtils.sendMeterOrCloud(deviceManageEntityList.get(0), 0, IP, port, ctx);
|
||||||
} else { |
// }
|
||||||
// 添加一个状态值,判断是否继续发送指令 update by ljf on 2020-08-07
|
// }
|
||||||
if (Constant.WEB_FLAG) { |
//
|
||||||
log.info("有指令下发退出定时采集DDC参数"); |
// @Override
|
||||||
num = 0; |
// public void channelInactive(ChannelHandlerContext ctx) throws Exception {
|
||||||
// 关闭连接
|
// Thread.sleep(500);
|
||||||
receiveStr = null; |
// receiveStr = null;
|
||||||
ctx.close(); |
// ctx.close();
|
||||||
} else { |
// log.info(ctx.channel().localAddress() + "退出链接!!");
|
||||||
// 封装发送电表工具方法 update by ljf on 2021-01-26
|
// }
|
||||||
SendOrderUtils.sendMeterOrCloud(deviceManageEntityList.get(num), num, IP, port, ctx); |
//
|
||||||
} |
// @Override
|
||||||
} |
// public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
|
||||||
ctx.flush(); |
// try {
|
||||||
} |
// ByteBuf buf = (ByteBuf) msg;
|
||||||
} |
// byte[] bytes = new byte[buf.readableBytes()];
|
||||||
|
// buf.readBytes(bytes);//复制内容到字节数组bytes
|
||||||
|
// buf.clear();
|
||||||
|
// log.info("获取到的值: " + ExchangeStringUtil.bytesToHexString(bytes));
|
||||||
|
// if (bytes.length <= 62) {
|
||||||
|
//// if (bytes.length <= 142) {
|
||||||
|
//// receiveStr = receiveStr.replace("null", "");
|
||||||
|
//// receiveStr = receiveStr + ExchangeStringUtil.bytesToHexString(bytes);//将接收到的数据转为字符串,此字符串就是客户端发送的字符串
|
||||||
|
//// log.info(ctx.channel().remoteAddress() + " " + ctx.channel().localAddress() + " 接受服务器数据:" + receiveStr + ",大小: " + receiveStr.length());
|
||||||
|
// receiveStr = receiveStr + ExchangeStringUtil.bytesToHexString(bytes);//将接收到的数据转为字符串,此字符串就是客户端发送的字符串
|
||||||
|
// receiveStr = receiveStr.replace("null", "");
|
||||||
|
// log.info("接受服务器数据:" + receiveStr + ",大小: " + receiveStr.length());
|
||||||
|
// }
|
||||||
|
// } catch (Exception e) {
|
||||||
|
// e.printStackTrace();
|
||||||
|
// } finally {
|
||||||
|
// ReferenceCountUtil.release(msg);
|
||||||
|
// }
|
||||||
|
// }
|
||||||
|
//
|
||||||
|
// @Override
|
||||||
|
// public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
|
||||||
|
// log.info("采集电表或者冷量计-数据读取接收完成: " + receiveStr);
|
||||||
|
// if ((receiveStr.length() == 36) && (deviceManageEntityList.get(num).getParamId() == 3)) {
|
||||||
|
//// analysisReceiveOrder485.analysisCloudOrder485(receiveStr); // 解析冷量计
|
||||||
|
// } else if ((receiveStr.length() == 62) && (deviceManageEntityList.get(num).getParamId() == 1)) {
|
||||||
|
//// 把receiveStr的"null"值去掉
|
||||||
|
//// a9fec2c71f9002fefefefe6839025007000068810643c3bb446c338d16c2b8
|
||||||
|
//// A9 FE C2 C7 1F 90 02 FE FE FE FE 68 39 02 50 07 00 00 68 81 06 43 C3 5B 38 6C 33 21 16 F8 12
|
||||||
|
// analysisReceiveOrder485.analysisMeterOrder485(receiveStr,"","","", null); // 解析电表
|
||||||
|
// }
|
||||||
|
// receiveStr = "";
|
||||||
|
// num = num + 1;
|
||||||
|
// Thread.sleep(600);
|
||||||
|
// if (num > size - 1) {
|
||||||
|
// num = 0;
|
||||||
|
// receiveStr = null;
|
||||||
|
// // 关闭连接
|
||||||
|
// ctx.close();
|
||||||
|
//// // 保持长连接,封装发送电表工具方法 update by ljf on 2021-01-26
|
||||||
|
//// SendOrderUtils.sendMeterOrder(deviceManageEntityList.get(num),num,IP,port,ctx);
|
||||||
|
// } else {
|
||||||
|
// // 添加一个状态值,判断是否继续发送指令 update by ljf on 2020-08-07
|
||||||
|
// if (Constant.WEB_FLAG) {
|
||||||
|
// log.info("有指令下发退出定时采集DDC参数");
|
||||||
|
// num = 0;
|
||||||
|
// // 关闭连接
|
||||||
|
// receiveStr = null;
|
||||||
|
// ctx.close();
|
||||||
|
// } else {
|
||||||
|
// // 封装发送电表工具方法 update by ljf on 2021-01-26
|
||||||
|
// SendOrderUtils.sendMeterOrCloud(deviceManageEntityList.get(num), num, IP, port, ctx);
|
||||||
|
// }
|
||||||
|
// }
|
||||||
|
// ctx.flush();
|
||||||
|
// }
|
||||||
|
//}
|
||||||
|
@ -1,44 +1,44 @@ |
|||||||
package com.mh.user.socket; |
//package com.mh.user.socket;
|
||||||
|
//
|
||||||
import com.mh.user.netty.NettyClient; |
//import com.mh.user.netty.NettyClient;
|
||||||
import com.mh.user.netty.NettyMeterClient; |
//import com.mh.user.netty.NettyMeterClient;
|
||||||
import lombok.extern.slf4j.Slf4j; |
//import lombok.extern.slf4j.Slf4j;
|
||||||
import org.junit.jupiter.api.Test; |
//import org.junit.jupiter.api.Test;
|
||||||
import org.springframework.boot.test.context.SpringBootTest; |
//import org.springframework.boot.test.context.SpringBootTest;
|
||||||
|
//
|
||||||
/** |
///**
|
||||||
* @author ljf |
// * @author ljf
|
||||||
* @title : |
// * @title :
|
||||||
* @description : |
// * @description :
|
||||||
* @updateTime 2020-05-13 |
// * @updateTime 2020-05-13
|
||||||
* @throws : |
// * @throws :
|
||||||
*/ |
// */
|
||||||
@Slf4j |
//@Slf4j
|
||||||
@SpringBootTest |
//@SpringBootTest
|
||||||
public class NettyClientTest { |
//public class NettyClientTest {
|
||||||
|
//
|
||||||
@Test |
// @Test
|
||||||
void Test() throws InterruptedException { |
// void Test() throws InterruptedException {
|
||||||
NettyClient nettyClient = new NettyClient(); |
// NettyClient nettyClient = new NettyClient();
|
||||||
nettyClient.connect(1008,"192.168.1.222"); |
// nettyClient.connect(1008,"192.168.1.222");
|
||||||
// Thread.sleep(5000);
|
//// Thread.sleep(5000);
|
||||||
// nettyClient.connect(10100,"192.168.1.222");
|
//// nettyClient.connect(10100,"192.168.1.222");
|
||||||
} |
// }
|
||||||
|
//
|
||||||
@Test |
// @Test
|
||||||
void Test1() throws InterruptedException { |
// void Test1() throws InterruptedException {
|
||||||
NettyClient nettyClient = new NettyClient(); |
// NettyClient nettyClient = new NettyClient();
|
||||||
nettyClient.connect(8080,"192.168.2.10"); |
// nettyClient.connect(8080,"192.168.2.10");
|
||||||
// Thread.sleep(5000);
|
//// Thread.sleep(5000);
|
||||||
// nettyClient.connect(10100,"192.168.1.222");
|
//// nettyClient.connect(10100,"192.168.1.222");
|
||||||
} |
// }
|
||||||
|
//
|
||||||
|
//
|
||||||
@Test |
// @Test
|
||||||
void testMeterChillers() throws InterruptedException { |
// void testMeterChillers() throws InterruptedException {
|
||||||
NettyMeterClient nettyMeterClient = new NettyMeterClient(); |
// NettyMeterClient nettyMeterClient = new NettyMeterClient();
|
||||||
nettyMeterClient.connect(8080,"192.168.2.10"); |
// nettyMeterClient.connect(8080,"192.168.2.10");
|
||||||
// nettyMeterClient.connect(9000,"192.168.2.54");
|
//// nettyMeterClient.connect(9000,"192.168.2.54");
|
||||||
} |
// }
|
||||||
|
//
|
||||||
} |
//}
|
||||||
|
@ -1,182 +1,182 @@ |
|||||||
package com.mh.user.socket; |
//package com.mh.user.socket;
|
||||||
|
//
|
||||||
import com.mh.user.utils.ExchangeStringUtil; |
//import com.mh.user.utils.ExchangeStringUtil;
|
||||||
import io.netty.bootstrap.ServerBootstrap; |
//import io.netty.bootstrap.ServerBootstrap;
|
||||||
import io.netty.buffer.ByteBuf; |
//import io.netty.buffer.ByteBuf;
|
||||||
import io.netty.buffer.Unpooled; |
//import io.netty.buffer.Unpooled;
|
||||||
import io.netty.channel.*; |
//import io.netty.channel.*;
|
||||||
import io.netty.channel.nio.NioEventLoopGroup; |
//import io.netty.channel.nio.NioEventLoopGroup;
|
||||||
import io.netty.channel.socket.SocketChannel; |
//import io.netty.channel.socket.SocketChannel;
|
||||||
import io.netty.channel.socket.nio.NioServerSocketChannel; |
//import io.netty.channel.socket.nio.NioServerSocketChannel;
|
||||||
import io.netty.handler.logging.LogLevel; |
//import io.netty.handler.logging.LogLevel;
|
||||||
import io.netty.handler.logging.LoggingHandler; |
//import io.netty.handler.logging.LoggingHandler;
|
||||||
import lombok.extern.slf4j.Slf4j; |
//import lombok.extern.slf4j.Slf4j;
|
||||||
import org.springframework.util.StringUtils; |
//import org.springframework.util.StringUtils;
|
||||||
|
//
|
||||||
import java.io.IOException; |
//import java.io.IOException;
|
||||||
import java.net.InetSocketAddress; |
//import java.net.InetSocketAddress;
|
||||||
|
//
|
||||||
|
//
|
||||||
/** |
///**
|
||||||
* @author ljf |
// * @author ljf
|
||||||
* @title :Netty |
// * @title :Netty
|
||||||
* @description :netty 使用 |
// * @description :netty 使用
|
||||||
* @updateTime 2020-04-21 |
// * @updateTime 2020-04-21
|
||||||
* @throws : |
// * @throws :
|
||||||
*/ |
// */
|
||||||
@Slf4j |
//@Slf4j
|
||||||
public class NettyEchoServer { |
//public class NettyEchoServer {
|
||||||
|
//
|
||||||
public void bind(int port) throws Exception { |
// public void bind(int port) throws Exception {
|
||||||
// accept线程组,用来接收连接
|
// // accept线程组,用来接收连接
|
||||||
EventLoopGroup bossGroup = new NioEventLoopGroup(1); |
// EventLoopGroup bossGroup = new NioEventLoopGroup(1);
|
||||||
// IO 线程组,用来处理业务逻辑
|
// // IO 线程组,用来处理业务逻辑
|
||||||
EventLoopGroup workerGroup = new NioEventLoopGroup(1); |
// EventLoopGroup workerGroup = new NioEventLoopGroup(1);
|
||||||
|
//
|
||||||
try { |
// try {
|
||||||
// 服务端启动引导
|
// // 服务端启动引导
|
||||||
ServerBootstrap serverBootstrap = new ServerBootstrap(); |
// ServerBootstrap serverBootstrap = new ServerBootstrap();
|
||||||
serverBootstrap.group(bossGroup,workerGroup) // 绑定两个线程
|
// serverBootstrap.group(bossGroup,workerGroup) // 绑定两个线程
|
||||||
.channel(NioServerSocketChannel.class) // 指定通道类型
|
// .channel(NioServerSocketChannel.class) // 指定通道类型
|
||||||
.option(ChannelOption.SO_BACKLOG, 1024) // 设置TCP连接的缓冲区
|
// .option(ChannelOption.SO_BACKLOG, 1024) // 设置TCP连接的缓冲区
|
||||||
.handler(new LoggingHandler(LogLevel.INFO)) // 设置日志级别
|
// .handler(new LoggingHandler(LogLevel.INFO)) // 设置日志级别
|
||||||
.childHandler(new ChannelInitializer<SocketChannel>() { |
// .childHandler(new ChannelInitializer<SocketChannel>() {
|
||||||
@Override |
// @Override
|
||||||
protected void initChannel(SocketChannel socketChannel) throws Exception { |
// protected void initChannel(SocketChannel socketChannel) throws Exception {
|
||||||
ChannelPipeline pipeline = socketChannel.pipeline(); // 获取处理器链
|
// ChannelPipeline pipeline = socketChannel.pipeline(); // 获取处理器链
|
||||||
pipeline.addLast(new EchoServerHandler()); // 添加新的事件处理器
|
// pipeline.addLast(new EchoServerHandler()); // 添加新的事件处理器
|
||||||
} |
// }
|
||||||
}); |
// });
|
||||||
// 通过bind启动服务
|
// // 通过bind启动服务
|
||||||
ChannelFuture f = serverBootstrap.bind(port).sync(); |
// ChannelFuture f = serverBootstrap.bind(port).sync();
|
||||||
// 阻塞主线程,知道网络服务被关闭
|
// // 阻塞主线程,知道网络服务被关闭
|
||||||
f.channel().closeFuture().sync(); |
// f.channel().closeFuture().sync();
|
||||||
} catch (Exception e){ |
// } catch (Exception e){
|
||||||
e.printStackTrace(); |
// e.printStackTrace();
|
||||||
} finally { |
// } finally {
|
||||||
workerGroup.shutdownGracefully(); |
// workerGroup.shutdownGracefully();
|
||||||
bossGroup.shutdownGracefully(); |
// bossGroup.shutdownGracefully();
|
||||||
} |
// }
|
||||||
} |
// }
|
||||||
|
//
|
||||||
static class EchoServerHandler extends ChannelHandlerAdapter { |
// static class EchoServerHandler extends ChannelHandlerAdapter {
|
||||||
|
//
|
||||||
// 每当从客户端收到新的数据时,这个方法会在收到消息时被调用
|
// // 每当从客户端收到新的数据时,这个方法会在收到消息时被调用
|
||||||
@Override |
// @Override
|
||||||
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { |
// public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
|
||||||
try { |
// try {
|
||||||
ByteBuf buf = (ByteBuf)msg; |
// ByteBuf buf = (ByteBuf)msg;
|
||||||
byte [] bytes = new byte[buf.readableBytes()]; |
// byte [] bytes = new byte[buf.readableBytes()];
|
||||||
buf.readBytes(bytes);//复制内容到字节数组bytes
|
// buf.readBytes(bytes);//复制内容到字节数组bytes
|
||||||
String receiveStr = ExchangeStringUtil.bytesToHexString(bytes);//将接收到的数据转为字符串,此字符串就是客户端发送的字符串
|
// String receiveStr = ExchangeStringUtil.bytesToHexString(bytes);//将接收到的数据转为字符串,此字符串就是客户端发送的字符串
|
||||||
log.info("接收到的数据: "+ receiveStr); |
// log.info("接收到的数据: "+ receiveStr);
|
||||||
//返回16进制到客户端
|
// //返回16进制到客户端
|
||||||
writeToClient(receiveStr,ctx,"测试"); |
// writeToClient(receiveStr,ctx,"测试");
|
||||||
} catch (Exception e) { |
// } catch (Exception e) {
|
||||||
// TODO Auto-generated catch block
|
// // TODO Auto-generated catch block
|
||||||
e.printStackTrace(); |
// e.printStackTrace();
|
||||||
} |
// }
|
||||||
// ctx.write(Unpooled.wrappedBuffer("Server message".getBytes()));
|
//// ctx.write(Unpooled.wrappedBuffer("Server message".getBytes()));
|
||||||
// ctx.fireChannelRead(msg);
|
//// ctx.fireChannelRead(msg);
|
||||||
} |
// }
|
||||||
|
//
|
||||||
// 数据读取完后被调用
|
// // 数据读取完后被调用
|
||||||
@Override |
// @Override
|
||||||
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { |
// public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
|
||||||
ctx.flush(); |
// ctx.flush();
|
||||||
} |
// }
|
||||||
|
//
|
||||||
// 当Netty由于IO错误或者处理器在处理事件时抛出的异常时被调用
|
// // 当Netty由于IO错误或者处理器在处理事件时抛出的异常时被调用
|
||||||
@Override |
// @Override
|
||||||
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { |
// public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
|
||||||
cause.printStackTrace(); |
// cause.printStackTrace();
|
||||||
ctx.close(); |
// ctx.close();
|
||||||
} |
// }
|
||||||
|
//
|
||||||
/** |
// /**
|
||||||
* 客户端与服务端第一次建立连接时 执行 |
// * 客户端与服务端第一次建立连接时 执行
|
||||||
* |
// *
|
||||||
* @param ctx |
// * @param ctx
|
||||||
* @throws Exception |
// * @throws Exception
|
||||||
*/ |
// */
|
||||||
@Override |
// @Override
|
||||||
public void channelActive(ChannelHandlerContext ctx) throws Exception, IOException |
// public void channelActive(ChannelHandlerContext ctx) throws Exception, IOException
|
||||||
{ |
// {
|
||||||
super.channelActive(ctx); |
// super.channelActive(ctx);
|
||||||
ctx.channel().read(); |
// ctx.channel().read();
|
||||||
InetSocketAddress ifsock = (InetSocketAddress) ctx.channel().remoteAddress(); |
// InetSocketAddress ifsock = (InetSocketAddress) ctx.channel().remoteAddress();
|
||||||
String clientIp = ifsock.getAddress().getHostAddress(); |
// String clientIp = ifsock.getAddress().getHostAddress();
|
||||||
//此处不能使用ctx.close(),否则客户端始终无法与服务端建立连接
|
// //此处不能使用ctx.close(),否则客户端始终无法与服务端建立连接
|
||||||
log.info("channelActive: "+clientIp + ctx.name()); |
// log.info("channelActive: "+clientIp + ctx.name());
|
||||||
} |
// }
|
||||||
|
//
|
||||||
/** |
// /**
|
||||||
* 客户端与服务端 断连时 执行 |
// * 客户端与服务端 断连时 执行
|
||||||
* |
// *
|
||||||
* @param ctx |
// * @param ctx
|
||||||
* @throws Exception |
// * @throws Exception
|
||||||
*/ |
// */
|
||||||
@Override |
// @Override
|
||||||
public void channelInactive(ChannelHandlerContext ctx) throws Exception, IOException |
// public void channelInactive(ChannelHandlerContext ctx) throws Exception, IOException
|
||||||
{ |
// {
|
||||||
super.channelInactive(ctx); |
// super.channelInactive(ctx);
|
||||||
InetSocketAddress ifsock = (InetSocketAddress) ctx.channel().remoteAddress(); |
// InetSocketAddress ifsock = (InetSocketAddress) ctx.channel().remoteAddress();
|
||||||
String clientIp = ifsock.getAddress().getHostAddress(); |
// String clientIp = ifsock.getAddress().getHostAddress();
|
||||||
ctx.close(); //断开连接时,必须关闭,否则造成资源浪费,并发量很大情况下可能造成宕机
|
// ctx.close(); //断开连接时,必须关闭,否则造成资源浪费,并发量很大情况下可能造成宕机
|
||||||
System.out.println("channelInactive:"+clientIp); |
// System.out.println("channelInactive:"+clientIp);
|
||||||
} |
// }
|
||||||
|
//
|
||||||
/** |
// /**
|
||||||
* 服务端当read超时, 会调用这个方法 |
// * 服务端当read超时, 会调用这个方法
|
||||||
* |
// *
|
||||||
* @param ctx |
// * @param ctx
|
||||||
* @param evt |
// * @param evt
|
||||||
* @throws Exception |
// * @throws Exception
|
||||||
*/ |
// */
|
||||||
@Override |
// @Override
|
||||||
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception, IOException |
// public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception, IOException
|
||||||
{ |
// {
|
||||||
super.userEventTriggered(ctx, evt); |
// super.userEventTriggered(ctx, evt);
|
||||||
InetSocketAddress ifsock = (InetSocketAddress) ctx.channel().remoteAddress(); |
// InetSocketAddress ifsock = (InetSocketAddress) ctx.channel().remoteAddress();
|
||||||
String clientIp = ifsock.getAddress().getHostAddress(); |
// String clientIp = ifsock.getAddress().getHostAddress();
|
||||||
ctx.close();//超时时断开连接
|
// ctx.close();//超时时断开连接
|
||||||
System.out.println("userEventTriggered:"+clientIp); |
// System.out.println("userEventTriggered:"+clientIp);
|
||||||
} |
// }
|
||||||
|
//
|
||||||
|
//
|
||||||
/** |
// /**
|
||||||
* 公用回写数据到客户端的方法 |
// * 公用回写数据到客户端的方法
|
||||||
* @param channel |
// * @param channel
|
||||||
* @param mark 用于打印/log的输出 |
// * @param mark 用于打印/log的输出
|
||||||
* <br>//channel.writeAndFlush(msg);//不行
|
// * <br>//channel.writeAndFlush(msg);//不行
|
||||||
* <br>//channel.writeAndFlush(receiveStr.getBytes());//不行
|
// * <br>//channel.writeAndFlush(receiveStr.getBytes());//不行
|
||||||
* <br>在netty里,进出的都是ByteBuf,楼主应确定服务端是否有对应的编码器,将字符串转化为ByteBuf |
// * <br>在netty里,进出的都是ByteBuf,楼主应确定服务端是否有对应的编码器,将字符串转化为ByteBuf
|
||||||
*/ |
// */
|
||||||
private void writeToClient(final String receiveStr, ChannelHandlerContext channel, final String mark) { |
// private void writeToClient(final String receiveStr, ChannelHandlerContext channel, final String mark) {
|
||||||
try { |
// try {
|
||||||
ByteBuf buff = Unpooled.buffer();//netty需要用ByteBuf传输
|
// ByteBuf buff = Unpooled.buffer();//netty需要用ByteBuf传输
|
||||||
buff.writeBytes(ExchangeStringUtil.hexStrToBinaryStr(receiveStr));//对接需要16进制
|
// buff.writeBytes(ExchangeStringUtil.hexStrToBinaryStr(receiveStr));//对接需要16进制
|
||||||
channel.writeAndFlush(buff).addListener(new ChannelFutureListener() { |
// channel.writeAndFlush(buff).addListener(new ChannelFutureListener() {
|
||||||
@Override |
// @Override
|
||||||
public void operationComplete(ChannelFuture future) throws Exception { |
// public void operationComplete(ChannelFuture future) throws Exception {
|
||||||
StringBuilder sb = new StringBuilder(""); |
// StringBuilder sb = new StringBuilder("");
|
||||||
if(!StringUtils.isEmpty(mark)){ |
// if(!StringUtils.isEmpty(mark)){
|
||||||
sb.append("【").append(mark).append("】"); |
// sb.append("【").append(mark).append("】");
|
||||||
} |
// }
|
||||||
if (future.isSuccess()) { |
// if (future.isSuccess()) {
|
||||||
System.out.println(sb.toString()+"回写成功"+receiveStr); |
// System.out.println(sb.toString()+"回写成功"+receiveStr);
|
||||||
log.info(sb.toString()+"回写成功"+receiveStr); |
// log.info(sb.toString()+"回写成功"+receiveStr);
|
||||||
} else { |
// } else {
|
||||||
System.out.println(sb.toString()+"回写失败"+receiveStr); |
// System.out.println(sb.toString()+"回写失败"+receiveStr);
|
||||||
log.error(sb.toString()+"回写失败"+receiveStr); |
// log.error(sb.toString()+"回写失败"+receiveStr);
|
||||||
} |
// }
|
||||||
} |
// }
|
||||||
}); |
// });
|
||||||
} catch (Exception e) { |
// } catch (Exception e) {
|
||||||
e.printStackTrace(); |
// e.printStackTrace();
|
||||||
System.out.println("调用通用writeToClient()异常"+e.getMessage()); |
// System.out.println("调用通用writeToClient()异常"+e.getMessage());
|
||||||
log.error("调用通用writeToClient()异常:",e); |
// log.error("调用通用writeToClient()异常:",e);
|
||||||
} |
// }
|
||||||
} |
// }
|
||||||
} |
// }
|
||||||
} |
//}
|
||||||
|
Loading…
Reference in new issue