autumn/jsowell-charger-web
1
0
Fork 0
mirror of https://codeup.aliyun.com/67c68d4e484ca2f0a13ac3c1/ydc/jsowell-charger-web.git synced 2026-04-20 02:55:04 +08:00
Code Issues Packages Projects Releases Wiki Activity

update 修复离线bug

Browse Source
This commit is contained in:
Guoqs
2026-03-21 09:20:58 +08:00
parent c9d7f505b6
commit 0e763d4fe2
9 changed files with 687 additions and 207 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
jsowell-netty/src/main/java/com/jsowell/netty/handler/yunkuaichong/AbstractYkcHandler.java
View File

@@ -67,6 +67,9 @@ public abstract class AbstractYkcHandler implements InitializingBean {
String redisKey = CacheConstants.PILE_LAST_CONNECTION + pileSn;
redisCache.setCacheObject(redisKey, DateUtils.getDateTime(), CacheConstants.cache_expire_time_30d);
redisCache.deleteObject(CacheConstants.PILE_PENDING_DISCONNECT + pileSn);
redisCache.deleteObject(CacheConstants.PILE_OFFLINE_CONFIRMED + pileSn);
// 保存桩号和channel的关系
PileChannelEntity.checkChannel(pileSn, ctx);
@@ -90,4 +93,4 @@ public abstract class AbstractYkcHandler implements InitializingBean {
return !result;
}
}
}

4
jsowell-netty/src/main/java/com/jsowell/netty/server/yunkuaichong/NettyServerChannelInitializer.java
View File

@@ -34,8 +34,8 @@ public class NettyServerChannelInitializer extends ChannelInitializer<SocketChan
// pipeline.addLast("frameDecoder", new YkcProtocolDecoder());
pipeline.addLast("decoder", new ByteArrayDecoder());
pipeline.addLast("encoder", new ByteArrayDecoder());
// 读超时时间设置为30s，0表示不监控
pipeline.addLast(new IdleStateHandler(30, 0, 0, TimeUnit.SECONDS));
// 读空闲tick设置为15s，连续3次空闲后再由handler执行关闭
pipeline.addLast(new IdleStateHandler(15, 0, 0, TimeUnit.SECONDS));
// pipeline.addLast("handler", nettyServerHandler);
pipeline.addLast(businessGroup, nettyServerHandler); // 消息先进入业务线程池
pipeline.addLast(echoServerHandler);

305
jsowell-netty/src/main/java/com/jsowell/netty/server/yunkuaichong/NettyServerHandler.java
View File

@@ -5,19 +5,22 @@ import com.jsowell.common.core.domain.ykc.YKCDataProtocol;
import com.jsowell.common.core.domain.ykc.YKCFrameTypeCode;
import com.jsowell.common.enums.ykc.PileChannelEntity;
import com.jsowell.common.util.BytesUtil;
import com.jsowell.common.util.StringUtils;
import com.jsowell.common.util.DateUtils;
import com.jsowell.common.util.YKCUtils;
import com.jsowell.netty.service.yunkuaichong.YKCBusinessService;
import com.jsowell.netty.service.yunkuaichong.impl.YKCBusinessServiceImpl;
import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelId;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.ReadTimeoutException;
import io.netty.util.AttributeKey;
import io.netty.util.ReferenceCountUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
@@ -34,6 +37,16 @@ import java.util.concurrent.ConcurrentHashMap;
@Component
public class NettyServerHandler extends ChannelInboundHandlerAdapter {
public static final AttributeKey<Integer> IDLE_COUNT = AttributeKey.valueOf("ykc_idle_count");
public static final AttributeKey<String> LAST_FRAME_TYPE = AttributeKey.valueOf("ykc_last_frame_type");
public static final AttributeKey<String> LAST_RECEIVE_AT = AttributeKey.valueOf("ykc_last_receive_at");
public static final AttributeKey<String> LAST_SERIAL_NUMBER = AttributeKey.valueOf("ykc_last_serial_number");
public static final AttributeKey<String> DISCONNECT_REASON = AttributeKey.valueOf("ykc_disconnect_reason");
private static final int IDLE_STRIKE_THRESHOLD = 3;
private static final String DISCONNECT_REASON_READER_IDLE_3X = "reader_idle_3x";
private static final String DISCONNECT_REASON_READ_TIMEOUT = "read_timeout";
@Resource
private YKCBusinessService ykcService;
@@ -42,9 +55,7 @@ public class NettyServerHandler extends ChannelInboundHandlerAdapter {
*/
private static final ConcurrentHashMap<ChannelId, ChannelHandlerContext> CHANNEL_MAP = new ConcurrentHashMap<>();
protected static final List<String> notPrintFrameTypeList = Lists.newArrayList("0x03"); // "0x03"
// private final YKCBusinessService ykcService = new YKCBusinessServiceImpl();
protected static final List<String> notPrintFrameTypeList = Lists.newArrayList("0x03");
/**
* 有客户端连接服务器会触发此函数
@@ -55,107 +66,42 @@ public class NettyServerHandler extends ChannelInboundHandlerAdapter {
InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress();
String clientIp = insocket.getAddress().getHostAddress();
int clientPort = insocket.getPort();
//获取连接通道唯一标识
ChannelId channelId = ctx.channel().id();
//如果map中不包含此连接，就保存连接
ctx.channel().attr(IDLE_COUNT).set(0);
if (CHANNEL_MAP.containsKey(channelId)) {
log.info("客户端【{}】是连接状态，连接通道数量: {}", channelId, CHANNEL_MAP.size());
} else {
//保存连接
CHANNEL_MAP.put(channelId, ctx);
log.info("客户端【{}】, 连接netty服务器【IP:{}, PORT:{}】, 连接通道数量: {}", channelId, clientIp, clientPort, CHANNEL_MAP.size());
}
}
/**
* 有客户端发消息会触发此函数
*/
// @Override
// public void channelRead(ChannelHandlerContext ctx, Object message) throws Exception {
// // log.info("channelRead-aClass:{}", message.getClass());
// // log.info("加载客户端报文channelRead=== channelId:" + ctx.channel().id() + ", msg:" + message);
// // 下面可以解析数据，保存数据，生成返回报文，将需要返回报文写入write函数
// YKCDataProtocol ykcDataProtocol = (YKCDataProtocol) message;
// byte[] msg = ykcDataProtocol.getBytes();
//
// // 获取帧类型
// byte[] frameTypeBytes = BytesUtil.copyBytes(msg, 5, 1);
// String frameType = YKCUtils.frameType2Str(frameTypeBytes);
//
// // 判断该帧类型是否为某请求帧的应答帧
// String requestFrameType = YKCFrameTypeCode.PileAnswersRelation.getRequestFrameType(frameType);
// // log.info("同步获取响应数据-判断该帧类型是否为某请求帧的应答帧, frameType:{}, requestFrameType:{}", frameType, requestFrameType);
// if (StringUtils.isNotBlank(requestFrameType)) {
// // 根据请求id，在集合中找到与外部线程通信的SyncPromise对象
// String msgId = ctx.channel().id().toString() + "_" + requestFrameType;
// // log.info("同步获取响应数据-收到消息, msgId:{}", msgId);
// SyncPromise syncPromise = RpcUtil.getSyncPromiseMap().get(msgId);
// if(syncPromise != null) {
// // 设置响应结果
// syncPromise.setRpcResult(msg);
// // 唤醒外部线程
// // log.info("同步获取响应数据-唤醒外部线程, SyncPromise:{}", JSON.toJSONString(syncPromise));
// syncPromise.wake();
// }
// }
//
// // 获取序列号域
// int serialNumber = BytesUtil.bytesToIntLittle(BytesUtil.copyBytes(msg, 2, 2));
//
// // 获取channel
// Channel channel = ctx.channel();
//
// // 心跳包0x03日志太多，造成日志文件过大，改为不打印
// if (!CollectionUtils.containsAny(notPrintFrameTypeList, frameType)) {
// log.info("【<<<<<平台收到消息<<<<<】channel:{}, 帧类型:{}, 帧名称:{}, 序列号域:{}, 报文:{}",
// channel.id(), frameType, YKCFrameTypeCode.getFrameTypeStr(frameType), serialNumber,
// BytesUtil.binary(msg, 16));
// }
//
// // 处理数据
// byte[] response = ykcService.process(msg, ctx);
// if (Objects.nonNull(response)) {
// // 响应客户端
// ByteBuf buffer = ctx.alloc().buffer().writeBytes(response);
// this.channelWrite(channel.id(), buffer);
// if (!CollectionUtils.containsAny(notPrintFrameTypeList, frameType)) {
// // 应答帧类型
// byte[] responseFrameTypeBytes = YKCFrameTypeCode.PlatformAnswersRelation.getResponseFrameTypeBytes(frameTypeBytes);
// String responseFrameType = YKCUtils.frameType2Str(responseFrameTypeBytes);
// log.info("【>>>>>平台响应消息>>>>>】channel:{}, 响应帧类型:{}, 响应帧名称:{}, 原帧类型:{}, 原帧名称:{}, 序列号域:{}, response:{}",
// channel.id(), responseFrameType, YKCFrameTypeCode.getFrameTypeStr(responseFrameType),
// frameType, YKCFrameTypeCode.getFrameTypeStr(frameType), serialNumber,
// BytesUtil.binary(response, 16));
// }
// }
// }
@Override
public void channelRead(ChannelHandlerContext ctx, Object message) throws Exception {
try {
YKCDataProtocol ykcDataProtocol = (YKCDataProtocol) message;
// 获取帧类型
byte[] frameTypeBytes = ykcDataProtocol.getFrameType();
String frameType = YKCUtils.frameType2Str(frameTypeBytes);
// 获取序列号域
int serialNumber = BytesUtil.bytesToIntLittle(ykcDataProtocol.getSerialNumber());
// 获取channel
Channel channel = ctx.channel();
// 心跳包0x03日志太多，造成日志文件过大，改为不打印
ctx.channel().attr(IDLE_COUNT).set(0);
ctx.channel().attr(LAST_FRAME_TYPE).set(frameType);
ctx.channel().attr(LAST_RECEIVE_AT).set(DateUtils.getDateTime());
ctx.channel().attr(LAST_SERIAL_NUMBER).set(String.valueOf(serialNumber));
ctx.channel().attr(DISCONNECT_REASON).set(null);
if (!CollectionUtils.containsAny(notPrintFrameTypeList, frameType)) {
log.info("【<<<<<平台收到消息<<<<<】channel:{}, 帧类型:{}, 帧名称:{}, 序列号域:{}, 报文:{}",
channel.id(), frameType, YKCFrameTypeCode.getFrameTypeStr(frameType), serialNumber,
BytesUtil.binary(ykcDataProtocol.getBytes(), 16));
}
// 处理数据
byte[] response = ykcService.process(ykcDataProtocol, ctx);
if (Objects.nonNull(response)) {
// 响应客户端
ByteBuf buffer = ctx.alloc().buffer().writeBytes(response);
// this.channelWrite(channel.id(), buffer);
super.channelRead(ctx, buffer);
if (!CollectionUtils.containsAny(notPrintFrameTypeList, frameType)) {
// 应答帧类型
byte[] responseFrameTypeBytes = YKCFrameTypeCode.PlatformAnswersRelation.getResponseFrameTypeBytes(frameTypeBytes);
String responseFrameType = YKCUtils.frameType2Str(responseFrameTypeBytes);
log.info("【>>>>>平台响应消息>>>>>】channel:{}, 响应帧类型:{}, 响应帧名称:{}, 原帧类型:{}, 原帧名称:{}, 序列号域:{}, response:{}",
@@ -171,32 +117,21 @@ public class NettyServerHandler extends ChannelInboundHandlerAdapter {
/**
* 批量发送信息
* @param pileSn
* @param ctx
* @param msg
*
* @param pileSn 桩编号
* @param msg 消息体
*/
private void channelWriteBatch(String pileSn, Object msg) {
// 获取该桩下的所有channel
List<ChannelHandlerContext> list = PileChannelEntity.pileMap.get(pileSn);
if(CollectionUtils.isEmpty(list)) {
if (CollectionUtils.isEmpty(list)) {
return;
}
// 批量写入
for (ChannelHandlerContext context : list) {
context.write(msg);
//刷新缓存区
context.flush();
}
// 如果通道不存在，则将连接删除
}
// @Override
// protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
// log.info("channelRead0=== channelId:" + ctx.channel().id() + ", msg:" + msg);
// }
/**
* 有客户端终止连接服务器会触发此函数
*/
@@ -205,33 +140,39 @@ public class NettyServerHandler extends ChannelInboundHandlerAdapter {
InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress();
String clientIp = insocket.getAddress().getHostAddress();
ChannelId channelId = ctx.channel().id();
//包含此客户端才去删除
String pileSn = PileChannelEntity.getPileSnByChannelId(channelId.asLongText());
String disconnectReason = defaultString(ctx.channel().attr(DISCONNECT_REASON).get(), "channel_inactive");
if (CHANNEL_MAP.containsKey(channelId)) {
ykcService.exit(ctx);
//删除连接
CHANNEL_MAP.remove(channelId);
}
log.info(
"云快充连接断开 pileSn:{}, channelId:{}, remoteAddress:{}, disconnectReason:{}, lastFrameType:{}, lastReceiveAt:{}, idleCount:{}, offlineConfirmed:{}, 连接通道数量:{}",
defaultString(pileSn),
channelId.asLongText(),
buildRemoteAddress(ctx),
disconnectReason,
defaultString(ctx.channel().attr(LAST_FRAME_TYPE).get()),
defaultString(ctx.channel().attr(LAST_RECEIVE_AT).get()),
getIdleCount(ctx),
false,
CHANNEL_MAP.size()
);
if (insocket != null) {
log.info("客户端【{}】, 退出netty服务器【IP:{}, PORT:{}】, 连接通道数量: {}", channelId, clientIp, insocket.getPort(), CHANNEL_MAP.size());
}
}
@Override
public void handlerAdded(ChannelHandlerContext ctx) throws Exception { // (2)
// Channel incoming = ctx.channel();
// log.info("handlerAdded: handler被添加到channel的pipeline connect:" + incoming.remoteAddress());
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
}
@Override
public void handlerRemoved(ChannelHandlerContext ctx) throws Exception { // (3)
// Channel incoming = ctx.channel();
// log.info("handlerRemoved: handler从channel的pipeline中移除 connect:" + incoming.remoteAddress());
// ChannelMapByEntity.removeChannel(incoming);
// ChannelMap.removeChannel(incoming);
public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
Channel channel = ctx.channel();
// log.info("channel:【{}】读数据完成", channel.id());
super.channelReadComplete(ctx);
}
@@ -239,7 +180,7 @@ public class NettyServerHandler extends ChannelInboundHandlerAdapter {
* 服务端给客户端发送消息
*
* @param channelId 连接通道唯一id
* @param msg 需要发送的消息内容
* @param msg 需要发送的消息内容
*/
public void channelWrite(ChannelId channelId, Object msg) throws Exception {
ChannelHandlerContext ctx = CHANNEL_MAP.get(channelId);
@@ -251,41 +192,64 @@ public class NettyServerHandler extends ChannelInboundHandlerAdapter {
log.info("服务端响应空的消息");
return;
}
//将客户端的信息直接返回写入ctx
ctx.write(msg);
//刷新缓存区
ctx.flush();
}
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
try {
String socketString = ctx.channel().remoteAddress().toString();
ChannelId channelId = ctx.channel().id();
String pileSn = PileChannelEntity.getPileSnByChannelId(channelId.asLongText());
if (evt instanceof IdleStateEvent) { // 超时事件
IdleStateEvent event = (IdleStateEvent) evt;
boolean flag = false;
if (event.state() == IdleState.READER_IDLE) { // 读
flag = true;
// log.error("Client-IP:【{}】, channelId:【{}】, pileSn:【{}】, READER_IDLE 读超时", socketString, channelId, pileSn);
} else if (event.state() == IdleState.WRITER_IDLE) { // 写
flag = true;
// log.error("Client-IP:【{}】, channelId:【{}】, pileSn:【{}】, WRITER_IDLE 写超时", socketString, channelId, pileSn);
} else if (event.state() == IdleState.ALL_IDLE) { // 全部
flag = true;
// log.error("Client-IP:【{}】, channelId:【{}】, pileSn:【{}】, ALL_IDLE 总超时", socketString, channelId, pileSn);
}
if (flag) {
ctx.channel().close();
// close(channelId, pileSn);
}
}
} finally {
if (evt instanceof ByteBuf) {
ReferenceCountUtil.release(evt);
}
if (!(evt instanceof IdleStateEvent)) {
super.userEventTriggered(ctx, evt);
return;
}
IdleStateEvent event = (IdleStateEvent) evt;
ChannelId channelId = ctx.channel().id();
String pileSn = PileChannelEntity.getPileSnByChannelId(channelId.asLongText());
if (event.state() == IdleState.READER_IDLE) {
int idleCount = getIdleCount(ctx) + 1;
ctx.channel().attr(IDLE_COUNT).set(idleCount);
if (idleCount < IDLE_STRIKE_THRESHOLD) {
log.warn(
"云快充连接读空闲告警 pileSn:{}, channelId:{}, remoteAddress:{}, disconnectReason:{}, lastFrameType:{}, lastReceiveAt:{}, idleCount:{}, lastSerialNumber:{}, pendingOffline:{}",
defaultString(pileSn),
channelId.asLongText(),
buildRemoteAddress(ctx),
defaultString(ctx.channel().attr(DISCONNECT_REASON).get()),
defaultString(ctx.channel().attr(LAST_FRAME_TYPE).get()),
defaultString(ctx.channel().attr(LAST_RECEIVE_AT).get()),
idleCount,
defaultString(ctx.channel().attr(LAST_SERIAL_NUMBER).get()),
false
);
return;
}
ctx.channel().attr(DISCONNECT_REASON).set(DISCONNECT_REASON_READER_IDLE_3X);
log.error(
"云快充连接连续读空闲关闭 pileSn:{}, channelId:{}, remoteAddress:{}, disconnectReason:{}, lastFrameType:{}, lastReceiveAt:{}, idleCount:{}, lastSerialNumber:{}, pendingOffline:{}",
defaultString(pileSn),
channelId.asLongText(),
buildRemoteAddress(ctx),
DISCONNECT_REASON_READER_IDLE_3X,
defaultString(ctx.channel().attr(LAST_FRAME_TYPE).get()),
defaultString(ctx.channel().attr(LAST_RECEIVE_AT).get()),
idleCount,
defaultString(ctx.channel().attr(LAST_SERIAL_NUMBER).get()),
true
);
ctx.close();
return;
}
log.warn(
"云快充连接空闲事件 pileSn:{}, channelId:{}, remoteAddress:{}, disconnectReason:{}, lastFrameType:{}, lastReceiveAt:{}, idleCount:{}, idleState:{}",
defaultString(pileSn),
channelId.asLongText(),
buildRemoteAddress(ctx),
defaultString(ctx.channel().attr(DISCONNECT_REASON).get()),
defaultString(ctx.channel().attr(LAST_FRAME_TYPE).get()),
defaultString(ctx.channel().attr(LAST_RECEIVE_AT).get()),
getIdleCount(ctx),
event.state()
);
}
/**
@@ -293,44 +257,39 @@ public class NettyServerHandler extends ChannelInboundHandlerAdapter {
*/
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
try {
ChannelId channelId = ctx.channel().id();
String channelIdShortText = channelId.asShortText();
String pileSn = PileChannelEntity.getPileSnByChannelId(channelIdShortText);
log.error("发生异常 channelId:{}, pileSn:{}", channelIdShortText, pileSn, cause);
cause.printStackTrace();
// 如果桩连到平台，在1分钟内没有发送数据过来，会报ReadTimeoutException异常
if (cause instanceof ReadTimeoutException) {
if (log.isTraceEnabled()) {
log.trace("Connection timeout 【{}】", ctx.channel().remoteAddress());
}
log.error("【{}】发生了错误, pileSn:【{}】此连接被关闭, 此时连通数量: {}", channelId, pileSn, CHANNEL_MAP.size());
// 删除连接
// PileChannelEntity.deleteChannel(pileSn, ctx);
ctx.channel().close();
}
} finally {
if (ctx.channel().isActive()) {
ctx.close();
}
}
ChannelId channelId = ctx.channel().id();
String pileSn = PileChannelEntity.getPileSnByChannelId(channelId.asLongText());
String disconnectReason = cause instanceof ReadTimeoutException ? DISCONNECT_REASON_READ_TIMEOUT : cause.getClass().getSimpleName();
ctx.channel().attr(DISCONNECT_REASON).set(disconnectReason);
log.error(
"云快充连接异常 pileSn:{}, channelId:{}, remoteAddress:{}, disconnectReason:{}, lastFrameType:{}, lastReceiveAt:{}, idleCount:{}, lastSerialNumber:{}",
defaultString(pileSn),
channelId.asLongText(),
buildRemoteAddress(ctx),
disconnectReason,
defaultString(ctx.channel().attr(LAST_FRAME_TYPE).get()),
defaultString(ctx.channel().attr(LAST_RECEIVE_AT).get()),
getIdleCount(ctx),
defaultString(ctx.channel().attr(LAST_SERIAL_NUMBER).get()),
cause
);
ctx.close();
}
private int getIdleCount(ChannelHandlerContext ctx) {
Integer idleCount = ctx.channel().attr(IDLE_COUNT).get();
return idleCount == null ? 0 : idleCount;
}
// 公共方法 关闭连接
private void closeConnection(String pileSn, ChannelHandlerContext ctx) {
Channel channel = ctx.channel();
ChannelId channelId = channel.id();
log.error("close方法-发生异常，关闭链接，channelId:{}, pileSn:{}", channelId.asShortText(), pileSn);
if (channel != null && !channel.isActive() && !channel.isOpen() && !channel.isWritable()) {
channel.close();
// 删除连接
CHANNEL_MAP.remove(channelId);
}
// 删除桩编号和channel的关系
if (StringUtils.isNotBlank(pileSn)) {
PileChannelEntity.removeByPileSn(pileSn);
}
private String buildRemoteAddress(ChannelHandlerContext ctx) {
return ctx.channel().remoteAddress() == null ? "" : ctx.channel().remoteAddress().toString();
}
private String defaultString(String value) {
return value == null ? "" : value;
}
private String defaultString(String value, String defaultValue) {
return value == null ? defaultValue : value;
}
}

205
jsowell-netty/src/main/java/com/jsowell/netty/service/yunkuaichong/impl/YKCBusinessServiceImpl.java
View File

@@ -1,29 +1,43 @@
package com.jsowell.netty.service.yunkuaichong.impl;
import com.jsowell.common.constant.CacheConstants;
import com.jsowell.common.core.domain.ykc.YKCDataProtocol;
import com.jsowell.common.core.domain.ykc.YKCFrameTypeCode;
import com.jsowell.common.core.redis.RedisCache;
import com.jsowell.common.enums.ykc.PileChannelEntity;
import com.jsowell.common.enums.ykc.PileConnectorDataBaseStatusEnum;
import com.jsowell.common.util.DateUtils;
import com.jsowell.common.util.StringUtils;
import com.jsowell.common.util.YKCUtils;
import com.jsowell.netty.factory.YKCOperateFactory;
import com.jsowell.netty.factory.YKCOperateFactoryV2;
import com.jsowell.netty.handler.yunkuaichong.AbstractYkcHandler;
import com.jsowell.netty.server.yunkuaichong.NettyServerHandler;
import com.jsowell.netty.service.yunkuaichong.YKCBusinessService;
import com.jsowell.netty.strategy.ykc.AbstractYkcStrategy;
import com.jsowell.pile.dto.SavePileMsgDTO;
import com.jsowell.pile.service.OrderBasicInfoService;
import com.jsowell.pile.service.PileConnectorInfoService;
import com.jsowell.pile.service.PileMsgRecordService;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import java.util.Date;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
@Slf4j
@Service
public class YKCBusinessServiceImpl implements YKCBusinessService {
private static final int OFFLINE_CONFIRM_SECONDS = 60;
private static final int OFFLINE_MARKER_TTL_SECONDS = 300;
private static final String OFFLINE_CONFIRMED_VALUE = "1";
@Autowired
private PileMsgRecordService pileMsgRecordService;
@@ -34,19 +48,21 @@ public class YKCBusinessServiceImpl implements YKCBusinessService {
private OrderBasicInfoService orderBasicInfoService;
@Autowired
private YKCOperateFactoryV2 ykcOperateFactoryV2; // 使用注解注入
private YKCOperateFactoryV2 ykcOperateFactoryV2;
@Autowired
private RedisCache redisCache;
@Resource(name = "scheduledExecutorService")
private ScheduledExecutorService scheduledExecutorService;
@Override
public byte[] process(byte[] msg, ChannelHandlerContext ctx) {
if (!YKCUtils.checkMsg(msg)) {
// 校验不通过，丢弃消息
return null;
}
YKCDataProtocol ykcDataProtocol = new YKCDataProtocol(msg);
// 获取帧类型
String frameType = YKCUtils.frameType2Str(ykcDataProtocol.getFrameType());
// 获取业务处理handler
// AbstractYkcHandler invokeStrategy = YKCOperateFactory.getInvokeStrategy(frameType);
AbstractYkcStrategy invokeStrategy = ykcOperateFactoryV2.getInvokeStrategy(frameType);
return invokeStrategy.supplyProcess(ykcDataProtocol, ctx);
}
@@ -54,53 +70,182 @@ public class YKCBusinessServiceImpl implements YKCBusinessService {
@Override
public byte[] process(YKCDataProtocol ykcDataProtocol, ChannelHandlerContext ctx) {
if (!YKCUtils.checkMsg(ykcDataProtocol)) {
// 校验不通过，丢弃消息
return null;
}
// 获取帧类型
String frameType = YKCUtils.frameType2Str(ykcDataProtocol.getFrameType());
// 获取业务处理handler
AbstractYkcHandler invokeStrategy = YKCOperateFactory.getInvokeStrategy(frameType); // 老逻辑
// AbstractYkcStrategy invokeStrategy = ykcOperateFactoryV2.getInvokeStrategy(frameType); // 新逻辑
AbstractYkcHandler invokeStrategy = YKCOperateFactory.getInvokeStrategy(frameType);
return invokeStrategy.supplyProcess(ykcDataProtocol, ctx);
}
@Override
public void exit(ChannelHandlerContext ctx) {
// 获取桩编号
String pileSn = PileChannelEntity.getPileSnByChannelId(ctx.channel().id().asLongText());
Channel channel = ctx.channel();
String channelId = channel.id().asLongText();
String pileSn = PileChannelEntity.getPileSnByChannelId(channelId);
if (StringUtils.isBlank(pileSn)) {
return;
}
log.info("充电桩退出:{}, 类型:主动断开链接, channelId:{}", pileSn, PileChannelEntity.getChannelByPileSn(pileSn).channel().id());
// 充电桩断开连接，所有枪口都设置为【离线】
pileConnectorInfoService.updateConnectorStatusByPileSn(pileSn, PileConnectorDataBaseStatusEnum.OFF_NETWORK.getValue());
String disconnectReason = defaultString(channel.attr(NettyServerHandler.DISCONNECT_REASON).get(), "channel_inactive");
String lastFrameType = defaultString(channel.attr(NettyServerHandler.LAST_FRAME_TYPE).get(), "");
String lastReceiveAt = defaultString(channel.attr(NettyServerHandler.LAST_RECEIVE_AT).get(), "");
String lastSerialNumber = defaultString(channel.attr(NettyServerHandler.LAST_SERIAL_NUMBER).get(), "");
String remoteAddress = buildRemoteAddress(ctx);
String occurredAt = DateUtils.getDateTime();
String pendingKey = CacheConstants.PILE_PENDING_DISCONNECT + pileSn;
String pendingValue = buildPendingValue(channelId, disconnectReason, occurredAt);
// 将此桩正在进行充电的订单状态改为 异常
orderBasicInfoService.updateOrderStatusAsAbnormal(pileSn);
PileChannelEntity.removeByPileSnAndChannelId(pileSn, channelId);
redisCache.setCacheObject(pendingKey, pendingValue, OFFLINE_MARKER_TTL_SECONDS);
// 记录充电桩退出msg
// 保存报文
String type = YKCFrameTypeCode.PILE_LOG_OUT.getCode() + "";
String jsonMsg = YKCFrameTypeCode.PILE_LOG_OUT.getValue() + ": 充电桩主动断开链接";
// pileMsgRecordService.save(pileSn, pileSn, type, jsonMsg, "");
log.warn(
"云快充断链进入待确认 pileSn:{}, channelId:{}, remoteAddress:{}, disconnectReason:{}, lastFrameType:{}, lastReceiveAt:{}, idleCount:{}, pendingOffline:{}, offlineConfirmed:{}",
pileSn,
channelId,
remoteAddress,
disconnectReason,
lastFrameType,
lastReceiveAt,
getIdleCount(ctx),
true,
false
);
savePileMsg(
pileSn,
YKCFrameTypeCode.PILE_LOG_OUT.getCode() + "",
YKCFrameTypeCode.PILE_LOG_OUT.getValue() + ": 连接断开待确认离线, reason=" + disconnectReason
+ ", channelId=" + channelId
+ ", lastFrameType=" + lastFrameType
+ ", lastReceiveAt=" + lastReceiveAt
+ ", lastSerialNumber=" + lastSerialNumber
);
scheduledExecutorService.schedule(
() -> confirmOffline(pileSn, channelId, disconnectReason, lastFrameType, lastReceiveAt, pendingValue, remoteAddress),
OFFLINE_CONFIRM_SECONDS,
TimeUnit.SECONDS
);
}
private void confirmOffline(String pileSn, String expectedChannelId, String disconnectReason, String lastFrameType,
String lastReceiveAt, String expectedPendingValue, String remoteAddress) {
String pendingKey = CacheConstants.PILE_PENDING_DISCONNECT + pileSn;
String offlineConfirmedKey = CacheConstants.PILE_OFFLINE_CONFIRMED + pileSn;
try {
String pendingValue = redisCache.getCacheObject(pendingKey);
if (StringUtils.isBlank(pendingValue)) {
return;
}
if (!StringUtils.equals(expectedPendingValue, pendingValue)) {
return;
}
ChannelHandlerContext currentCtx = PileChannelEntity.getChannelByPileSn(pileSn);
if (currentCtx != null && currentCtx.channel() != null) {
String currentChannelId = currentCtx.channel().id().asLongText();
if (!StringUtils.equals(currentChannelId, expectedChannelId)) {
redisCache.deleteObject(pendingKey);
log.info(
"云快充宽限期内连接恢复 pileSn:{}, channelId:{}, remoteAddress:{}, disconnectReason:{}, lastFrameType:{}, lastReceiveAt:{}, pendingOffline:{}, offlineConfirmed:{}",
pileSn,
currentChannelId,
buildRemoteAddress(currentCtx),
disconnectReason,
lastFrameType,
lastReceiveAt,
false,
false
);
return;
}
}
String lastConnectionTime = redisCache.getCacheObject(CacheConstants.PILE_LAST_CONNECTION + pileSn);
if (isRecentCommunication(lastConnectionTime)) {
redisCache.deleteObject(pendingKey);
log.info(
"云快充宽限期内通信恢复 pileSn:{}, channelId:{}, remoteAddress:{}, disconnectReason:{}, lastFrameType:{}, lastReceiveAt:{}, pendingOffline:{}, offlineConfirmed:{}",
pileSn,
expectedChannelId,
remoteAddress,
disconnectReason,
lastFrameType,
lastReceiveAt,
false,
false
);
return;
}
pileConnectorInfoService.updateConnectorStatusByPileSn(pileSn, PileConnectorDataBaseStatusEnum.OFF_NETWORK.getValue());
orderBasicInfoService.updateOrderStatusAsAbnormal(pileSn);
redisCache.setCacheObject(offlineConfirmedKey, OFFLINE_CONFIRMED_VALUE, OFFLINE_MARKER_TTL_SECONDS);
redisCache.deleteObject(pendingKey);
savePileMsg(
pileSn,
YKCFrameTypeCode.PILE_LOG_OUT.getCode() + "",
YKCFrameTypeCode.PILE_LOG_OUT.getValue() + ": 正式离线确认, reason=" + disconnectReason
+ ", channelId=" + expectedChannelId
+ ", lastFrameType=" + lastFrameType
+ ", lastReceiveAt=" + lastReceiveAt
);
log.error(
"云快充正式离线确认 pileSn:{}, channelId:{}, remoteAddress:{}, disconnectReason:{}, lastFrameType:{}, lastReceiveAt:{}, pendingOffline:{}, offlineConfirmed:{}",
pileSn,
expectedChannelId,
remoteAddress,
disconnectReason,
lastFrameType,
lastReceiveAt,
false,
true
);
} catch (Exception e) {
log.error("云快充离线确认任务执行异常 pileSn:{}, channelId:{}", pileSn, expectedChannelId, e);
}
}
private boolean isRecentCommunication(String lastConnectionTime) {
if (StringUtils.isBlank(lastConnectionTime)) {
return false;
}
Date lastConnectionDate = DateUtils.parseDate(lastConnectionTime);
if (lastConnectionDate == null) {
return false;
}
long diffMillis = System.currentTimeMillis() - lastConnectionDate.getTime();
return diffMillis >= 0 && diffMillis < OFFLINE_CONFIRM_SECONDS * 1000L;
}
private void savePileMsg(String pileSn, String frameType, String jsonMsg) {
SavePileMsgDTO dto = SavePileMsgDTO.builder()
.pileSn(pileSn)
.connectorCode("")
.transactionCode("")
.frameType(type)
.frameType(frameType)
.jsonMsg(jsonMsg)
.originalMsg(jsonMsg)
.build();
pileMsgRecordService.save(dto);
// 删除连接
// PileChannelEntity.deleteChannel(pileSn, ctx);
// 删除桩编号和channel的关系
// PileChannelEntity.removeByPileSn(pileSn);
}
private String buildPendingValue(String channelId, String reason, String occurredAt) {
return channelId + "|" + reason + "|" + occurredAt;
}
private int getIdleCount(ChannelHandlerContext ctx) {
Integer idleCount = ctx.channel().attr(NettyServerHandler.IDLE_COUNT).get();
return idleCount == null ? 0 : idleCount;
}
private String buildRemoteAddress(ChannelHandlerContext ctx) {
return ctx.channel().remoteAddress() == null ? "" : ctx.channel().remoteAddress().toString();
}
private String defaultString(String value, String defaultValue) {
return StringUtils.isBlank(value) ? defaultValue : value;
}
}